<!DOCTYPE html>
<html lang="zh-CN">
<head>
  <meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=2">
<meta name="theme-color" content="#222">
<meta name="generator" content="Hexo 4.2.1">
  <link rel="apple-touch-icon" sizes="180x180" href="/images/apple-touch-icon-next.png">
  <link rel="icon" type="image/png" sizes="32x32" href="/images/dute_favicon_32x32.png">
  <link rel="icon" type="image/png" sizes="16x16" href="/images/dute_favicon_16x16.png">
  <link rel="mask-icon" href="/images/logo.svg" color="#222">
  <link rel="manifest" href="/images/manifest.json">
  <meta name="msapplication-config" content="/images/browserconfig.xml">
  <meta http-equiv="Cache-Control" content="no-transform">
  <meta http-equiv="Cache-Control" content="no-siteapp">
  <meta name="google-site-verification" content="mpI5dkydstZXl6UcDCppqktXK0bbvqdZ6LkZ3KNk4Iw">
  <meta name="baidu-site-verification" content="code-a1LksZX2Ds">

<link rel="stylesheet" href="/css/main.css">


<link rel="stylesheet" href="/lib/font-awesome/css/font-awesome.min.css">
  <link rel="stylesheet" href="//cdn.jsdelivr.net/gh/fancyapps/fancybox@3/dist/jquery.fancybox.min.css">

<script id="hexo-configurations">
    var NexT = window.NexT || {};
    var CONFIG = {"hostname":"whitestore.top","root":"/","scheme":"Gemini","version":"7.8.0","exturl":true,"sidebar":{"position":"left","display":"post","padding":18,"offset":12,"onmobile":false},"copycode":{"enable":true,"show_result":false,"style":null},"back2top":{"enable":true,"sidebar":true,"scrollpercent":true},"bookmark":{"enable":false,"color":"#222","save":"auto"},"fancybox":true,"mediumzoom":false,"lazyload":false,"pangu":false,"comments":{"style":"tabs","active":null,"storage":true,"lazyload":false,"nav":null},"algolia":{"hits":{"per_page":10},"labels":{"input_placeholder":"Search for Posts","hits_empty":"We didn't find any results for the search: ${query}","hits_stats":"${hits} results found in ${time} ms"}},"localsearch":{"enable":true,"trigger":"auto","top_n_per_article":1,"unescape":false,"preload":false},"motion":{"enable":true,"async":false,"transition":{"post_block":"fadeIn","post_header":"slideDownIn","post_body":"slideDownIn","coll_header":"slideLeftIn","sidebar":"slideUpIn"}},"path":"search.xml"};
  </script>

  <meta name="description" content="学习和使用RocketMq">
<meta property="og:type" content="article">
<meta property="og:title" content="【RocketMq】商用RocketMq和开源RocketMq的兼容问题解决方案">
<meta property="og:url" content="https://whitestore.top/2023/03/11/shangyong/index.html">
<meta property="og:site_name" content="爱看书的阿东">
<meta property="og:description" content="学习和使用RocketMq">
<meta property="og:locale" content="zh_CN">
<meta property="og:image" content="https://adong-picture.oss-cn-shenzhen.aliyuncs.com/adong/image-20221229195508260.png">
<meta property="og:image" content="https://adong-picture.oss-cn-shenzhen.aliyuncs.com/adong/20221229175755.png">
<meta property="og:image" content="https://adong-picture.oss-cn-shenzhen.aliyuncs.com/adong/20221229182359.png">
<meta property="article:published_time" content="2023-03-10T23:21:11.000Z">
<meta property="article:modified_time" content="2023-06-15T13:10:39.213Z">
<meta property="article:author" content="阿东">
<meta property="article:tag" content="RocketMq学习">
<meta name="twitter:card" content="summary">
<meta name="twitter:image" content="https://adong-picture.oss-cn-shenzhen.aliyuncs.com/adong/image-20221229195508260.png">

<link rel="canonical" href="https://whitestore.top/2023/03/11/shangyong/">


<script id="page-configurations">
  // https://hexo.io/docs/variables.html
  CONFIG.page = {
    sidebar: "",
    isHome : false,
    isPost : true,
    lang   : 'zh-CN'
  };
</script>

  <title>【RocketMq】商用RocketMq和开源RocketMq的兼容问题解决方案 | 爱看书的阿东</title>
  






  <noscript>
  <style>
  .use-motion .brand,
  .use-motion .menu-item,
  .sidebar-inner,
  .use-motion .post-block,
  .use-motion .pagination,
  .use-motion .comments,
  .use-motion .post-header,
  .use-motion .post-body,
  .use-motion .collection-header { opacity: initial; }

  .use-motion .site-title,
  .use-motion .site-subtitle {
    opacity: initial;
    top: initial;
  }

  .use-motion .logo-line-before i { left: initial; }
  .use-motion .logo-line-after i { right: initial; }
  </style>
</noscript>

<link rel="alternate" href="/atom.xml" title="爱看书的阿东" type="application/atom+xml">
</head>

<body itemscope itemtype="http://schema.org/WebPage">
  <div class="container use-motion">
    <div class="headband"></div>

    <header class="header" itemscope itemtype="http://schema.org/WPHeader">
      <div class="header-inner"><div class="site-brand-container">
  <div class="site-nav-toggle">
    <div class="toggle" aria-label="切换导航栏">
      <span class="toggle-line toggle-line-first"></span>
      <span class="toggle-line toggle-line-middle"></span>
      <span class="toggle-line toggle-line-last"></span>
    </div>
  </div>

  <div class="site-meta">

    <a href="/" class="brand" rel="start">
      <span class="logo-line-before"><i></i></span>
      <h1 class="site-title">爱看书的阿东</h1>
      <span class="logo-line-after"><i></i></span>
    </a>
      <p class="site-subtitle" itemprop="description">赐他一块白色石头，石头上写着新名</p>
  </div>

  <div class="site-nav-right">
    <div class="toggle popup-trigger">
        <i class="fa fa-search fa-fw fa-lg"></i>
    </div>
  </div>
</div>




<nav class="site-nav">
  <ul id="menu" class="menu">
        <li class="menu-item menu-item-home">

    <a href="/" rel="section"><i class="fa fa-fw fa-home"></i>首页</a>

  </li>
        <li class="menu-item menu-item-tags">

    <a href="/tags/" rel="section"><i class="fa fa-fw fa-tags"></i>标签</a>

  </li>
        <li class="menu-item menu-item-categories">

    <a href="/categories/" rel="section"><i class="fa fa-fw fa-th"></i>分类</a>

  </li>
        <li class="menu-item menu-item-archives">

    <a href="/archives/" rel="section"><i class="fa fa-fw fa-archive"></i>归档</a>

  </li>
        <li class="menu-item menu-item-sitemap">

    <a href="/sitemap.xml" rel="section"><i class="fa fa-fw fa-sitemap"></i>站点地图</a>

  </li>
      <li class="menu-item menu-item-search">
        <a role="button" class="popup-trigger"><i class="fa fa-search fa-fw"></i>搜索
        </a>
      </li>
  </ul>
</nav>



  <div class="search-pop-overlay">
    <div class="popup search-popup">
        <div class="search-header">
  <span class="search-icon">
    <i class="fa fa-search"></i>
  </span>
  <div class="search-input-container">
    <input autocomplete="off" autocapitalize="off"
           placeholder="搜索..." spellcheck="false"
           type="search" class="search-input">
  </div>
  <span class="popup-btn-close">
    <i class="fa fa-times-circle"></i>
  </span>
</div>
<div id="search-result">
  <div id="no-result">
    <i class="fa fa-spinner fa-pulse fa-5x fa-fw"></i>
  </div>
</div>

    </div>
  </div>

</div>
    </header>

    

  <span class="exturl github-corner" data-url="aHR0cHM6Ly9naXRodWIuY29tL2xhenlUaW1lcw==" title="Follow me on GitHub" aria-label="Follow me on GitHub"><svg width="80" height="80" viewBox="0 0 250 250" aria-hidden="true"><path d="M0,0 L115,115 L130,115 L142,142 L250,250 L250,0 Z"></path><path d="M128.3,109.0 C113.8,99.7 119.0,89.6 119.0,89.6 C122.0,82.7 120.5,78.6 120.5,78.6 C119.2,72.0 123.4,76.3 123.4,76.3 C127.3,80.9 125.5,87.3 125.5,87.3 C122.9,97.6 130.6,101.9 134.4,103.2" fill="currentColor" style="transform-origin: 130px 106px;" class="octo-arm"></path><path d="M115.0,115.0 C114.9,115.1 118.7,116.5 119.8,115.4 L133.7,101.6 C136.9,99.2 139.9,98.4 142.2,98.6 C133.8,88.0 127.5,74.4 143.8,58.0 C148.5,53.4 154.0,51.2 159.7,51.0 C160.3,49.4 163.2,43.6 171.4,40.1 C171.4,40.1 176.1,42.5 178.8,56.2 C183.1,58.6 187.2,61.8 190.9,65.4 C194.5,69.0 197.7,73.2 200.1,77.6 C213.8,80.2 216.3,84.9 216.3,84.9 C212.7,93.1 206.9,96.0 205.4,96.6 C205.1,102.4 203.0,107.8 198.3,112.5 C181.9,128.9 168.3,122.5 157.7,114.1 C157.9,116.9 156.7,120.9 152.7,124.9 L141.0,136.5 C139.8,137.7 141.6,141.9 141.8,141.8 Z" fill="currentColor" class="octo-body"></path></svg></span>


    <main class="main">
      <div class="main-inner">
        <div class="content-wrap">
          

          <div class="content post posts-expand">
            

    
  
  
  <article itemscope itemtype="http://schema.org/Article" class="post-block" lang="zh-CN">
    <link itemprop="mainEntityOfPage" href="https://whitestore.top/2023/03/11/shangyong/">

    <span hidden itemprop="author" itemscope itemtype="http://schema.org/Person">
      <meta itemprop="image" content="/images/avatar.gif">
      <meta itemprop="name" content="阿东">
      <meta itemprop="description" content="随遇而安">
    </span>

    <span hidden itemprop="publisher" itemscope itemtype="http://schema.org/Organization">
      <meta itemprop="name" content="爱看书的阿东">
    </span>
      <header class="post-header">
        <h1 class="post-title" itemprop="name headline">
          【RocketMq】商用RocketMq和开源RocketMq的兼容问题解决方案
        </h1>

        <div class="post-meta">
            <span class="post-meta-item">
              <span class="post-meta-item-icon">
                <i class="fa fa-calendar-o"></i>
              </span>
              <span class="post-meta-item-text">发表于</span>

              <time title="创建时间：2023-03-11 07:21:11" itemprop="dateCreated datePublished" datetime="2023-03-11T07:21:11+08:00">2023-03-11</time>
            </span>
              <span class="post-meta-item">
                <span class="post-meta-item-icon">
                  <i class="fa fa-calendar-check-o"></i>
                </span>
                <span class="post-meta-item-text">更新于</span>
                <time title="修改时间：2023-06-15 21:10:39" itemprop="dateModified" datetime="2023-06-15T21:10:39+08:00">2023-06-15</time>
              </span>
            <span class="post-meta-item">
              <span class="post-meta-item-icon">
                <i class="fa fa-folder-o"></i>
              </span>
              <span class="post-meta-item-text">分类于</span>
                <span itemprop="about" itemscope itemtype="http://schema.org/Thing">
                  <a href="/categories/RocketMq/" itemprop="url" rel="index"><span itemprop="name">RocketMq</span></a>
                </span>
            </span>

          
            <span class="post-meta-item" title="阅读次数" id="busuanzi_container_page_pv" style="display: none;">
              <span class="post-meta-item-icon">
                <i class="fa fa-eye"></i>
              </span>
              <span class="post-meta-item-text">阅读次数：</span>
              <span id="busuanzi_value_page_pv"></span>
            </span>
  
  <span class="post-meta-item">
    
      <span class="post-meta-item-icon">
        <i class="fa fa-comment-o"></i>
      </span>
      <span class="post-meta-item-text">Valine：</span>
    
    <a title="valine" href="/2023/03/11/shangyong/#valine-comments" itemprop="discussionUrl">
      <span class="post-comments-count valine-comment-count" data-xid="/2023/03/11/shangyong/" itemprop="commentCount"></span>
    </a>
  </span>
  
  <br>
            <span class="post-meta-item" title="本文字数">
              <span class="post-meta-item-icon">
                <i class="fa fa-file-word-o"></i>
              </span>
                <span class="post-meta-item-text">本文字数：</span>
              <span>25k</span>
            </span>
            <span class="post-meta-item" title="阅读时长">
              <span class="post-meta-item-icon">
                <i class="fa fa-clock-o"></i>
              </span>
                <span class="post-meta-item-text">阅读时长 &asymp;</span>
              <span>23 分钟</span>
            </span>
            <div class="post-description">学习和使用RocketMq</div>

        </div>
      </header>

    
    
    
    <div class="post-body" itemprop="articleBody">

      
        <h1 id="引言"><a href="#引言" class="headerlink" title="引言"></a>引言</h1><p>在阿里云的官方网站提供了RocketMq的商用版本，但是个人在项目应用上发现和SpirngBoot以及Spring Cloud（Alibaba）等开源的RocketMQ依赖虽然可以正常兼容，但是依然出现了注解失效、启动报错，商用和开源版本的不兼容导致部分代码要重复编写的蛋疼问题。</p>
<p>这样的兼容问题不是简单加个SDK依赖，切换到商用配置就可以直接使用的（因为个人起初真就是这么想），为了避免后面再遇到这种<strong>奇葩</strong>的开发测试用开源RocketMq，生产环境需要使用商用集群的RocketMq的混合配置的业务场景，个人花了小半天时间熟读阿里云的接入文档，加上各种尝试和测试，总结出一套可以快速使用的兼容模板方案。</p>
<p>如果不了解阿里云商用RocketMq，可以看最后一个大节的【阿里云商用RocketMq介绍】介绍。个人的兼容方案灵感来自于官方提供的这个DEMO项目：<span class="exturl" data-url="aHR0cHM6Ly9naXRodWIuY29tL0FsaXdhcmVNUS9tcS1kZW1vL3RyZWUvbWFzdGVyL3NwcmluZ2Jvb3QvamF2YS1zcHJpbmdib290LWRlbW8=" title="https://github.com/AliwareMQ/mq-demo/tree/master/springboot/java-springboot-demo">springboot/java-springboot-demo<i class="fa fa-external-link"></i></span>。</p>
<p>注意本方案是基于<strong>SpringBoot2.X</strong>和 <strong>Spring cloud Alibaba</strong> 的两个项目环境构建项目基础，在SpringBoot上只做了生产者的配置，而在Spring Cloud Alibaba的Nacos上进行了生产者和消费者的完整兼容方案。</p>
<p>最后注意兼容集成的版本为商用RocketMq使用<strong>4.x</strong>版本，最近新出的5.X 的版本并<strong>未进行测试</strong>，不保证正常使用。</p>
<a id="more"></a>

<h1 id="兼容关键点"><a href="#兼容关键点" class="headerlink" title="兼容关键点"></a>兼容关键点</h1><ol>
<li>在沿用SpringBoot的YML基础配置基础上实现商用和开源模式的兼容。</li>
<li>商用RocketMq需要使用官方提供的依赖包，依赖包可以正常兼容SpringBoot等依赖。</li>
<li>集成之后便于开发和扩展，并且易于其他开发人员理解。</li>
<li>两种模式之间互相不会产生干扰。</li>
</ol>
<h1 id="SpringBoot2-X-兼容"><a href="#SpringBoot2-X-兼容" class="headerlink" title="SpringBoot2.X 兼容"></a>SpringBoot2.X 兼容</h1><p>下面先介绍<strong>SpringBoot项目</strong>的兼容。</p>
<h2 id="SpringBoot项目兼容"><a href="#SpringBoot项目兼容" class="headerlink" title="SpringBoot项目兼容"></a>SpringBoot项目兼容</h2><h3 id="开源版本"><a href="#开源版本" class="headerlink" title="开源版本"></a>开源版本</h3><p>首先我们观察YAML文件，对于<strong>开源版本的RocketMq</strong>设置，单机版本可以直接配置一个ip和端口即可，如果是集群则用分号隔开多个NameServer的连接IP地址（NameServ独立部署，内部进行自动同步 ）。</p>
<figure class="highlight yaml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">#消息队列</span></span><br><span class="line"><span class="attr">rocketmq:</span></span><br><span class="line">  <span class="comment"># 自定义属性，作用下文将会进行解释</span></span><br><span class="line">  <span class="attr">use-aliyun-rocketSever:</span> <span class="literal">false</span></span><br><span class="line">  <span class="attr">name-server:</span> <span class="number">192.168</span><span class="number">.58</span><span class="number">.128</span><span class="string">:9876</span> <span class="comment"># 192.168.244.128:9876;192.168.244.129:9876;</span></span><br><span class="line">  <span class="attr">producer:</span></span><br><span class="line">    <span class="attr">group:</span> <span class="string">testGroup</span></span><br><span class="line">    <span class="comment"># 本地开发不使用商业版，可以不配置</span></span><br><span class="line">    <span class="attr">secret-key:</span> <span class="string">NONE</span></span><br><span class="line">    <span class="comment"># 本地开发不使用商业版，可以不配置</span></span><br><span class="line">    <span class="attr">access-key:</span> <span class="string">NONE</span></span><br><span class="line">    <span class="comment"># 商用版本请求超时时间，开源版本不使用此参数</span></span><br><span class="line">    <span class="attr">timeoutMillis:</span> <span class="string">NONE</span></span><br></pre></td></tr></table></figure>

<p>SpringBoot中集成开源的RocketMq非常简单，只需要一个依赖就可以自动完成相关准备：</p>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>org.apache.rocketmq<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>rocketmq-spring-boot<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">version</span>&gt;</span>2.2.2<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br></pre></td></tr></table></figure>

<p>具体的使用通常为封装或者直接使用<code>RocketMqTemplate</code>：</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br></pre></td><td class="code"><pre><span class="line"></span><br><span class="line"><span class="meta">@Autowired</span></span><br><span class="line"><span class="keyword">private</span> RocketMQTemplate mqTemplate;</span><br><span class="line"></span><br><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">sendRocketMqUniqueTextMessage</span><span class="params">(String topic, String tag, String queueMsg)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">if</span> (StringUtils.isNotEmpty(topic) &amp;&amp; StringUtils.isNotEmpty(tag) &amp;&amp; StringUtils.isNotEmpty(queueMsg)) &#123;</span><br><span class="line">            String queueName = topic + <span class="string">":"</span> + tag;</span><br><span class="line">            <span class="comment">//封装消息，分配唯一id</span></span><br><span class="line">            MessageData messageData = <span class="keyword">new</span> MessageData();</span><br><span class="line">            messageData.setMsgId(IdUtil.randomUUID());</span><br><span class="line">            messageData.setMsgContent(queueMsg);</span><br><span class="line">            queueMsg = JSON.toJSONString(messageData);</span><br><span class="line">            log.info(<span class="string">"线程：&#123;&#125;，向队列：&#123;&#125;，发送消息：&#123;&#125;"</span>, Thread.currentThread()</span><br><span class="line">                    .getName(), queueName, queueMsg);</span><br><span class="line">            <span class="keyword">try</span> &#123;</span><br><span class="line">                mqTemplate.syncSend(queueName, queueMsg);</span><br><span class="line">            &#125; <span class="keyword">catch</span> (Exception e) &#123;</span><br><span class="line">                log.info(<span class="string">"向队列：&#123;&#125;，发送消息出现异常：&#123;&#125;"</span>, queueName, queueMsg);</span><br><span class="line">                <span class="comment">//出现异常，保存异常信息到数据库</span></span><br><span class="line">                SaMqMessageFail saMqMessageFail = <span class="keyword">new</span> SaMqMessageFail();</span><br><span class="line">                <span class="comment">// 封装失败消息，调用失效处理Service将失败发送请求入库，或许通过其他方法重试</span></span><br><span class="line">                saMqMessageFailService.insert(saMqMessageFail);</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br></pre></td></tr></table></figure>

<p>开源版本的SpringBoot集成RocketMq就是如此简单。</p>
<h3 id="商用版本"><a href="#商用版本" class="headerlink" title="商用版本"></a>商用版本</h3><p>商用版本RocketMq我们使用<span class="exturl" data-url="aHR0cHM6Ly9oZWxwLmFsaXl1bi5jb20vZG9jdW1lbnRfZGV0YWlsLzI5NTQ0Lmh0bWw=" title="https://help.aliyun.com/document_detail/29544.html">商业版TCP协议SDK（推荐）<i class="fa fa-external-link"></i></span>，注意这里用的是<strong>4.X版本</strong>。商用版本的YAML配置和开源版本<strong>显式配置</strong>是一样的，但是需要注意参数<code>use-aliyun-rocketSever=true</code>，并且<code>secretKey</code>和<code>accessKey</code>以及<code>name-server</code>都需要配置为阿里云提供的配置，最后设置消息发送超时时间<code>timeoutMillis</code>设置合理时间（单位为毫秒），便于排查问题和防止线程长期占用：</p>
<figure class="highlight yaml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">#消息队列</span></span><br><span class="line"><span class="attr">rocketmq:</span></span><br><span class="line">  <span class="attr">use-aliyun-rocketSever:</span> <span class="literal">true</span></span><br><span class="line">  <span class="comment"># 使用阿里云提供的endpoint</span></span><br><span class="line">  <span class="attr">name-server:</span> <span class="string">http://xxxxx.aliyuncs.com:8080</span></span><br><span class="line">  <span class="attr">producer:</span></span><br><span class="line">    <span class="attr">group:</span> <span class="string">testGroup</span></span><br><span class="line">    <span class="attr">secret-key:</span> <span class="string">xxx</span></span><br><span class="line">    <span class="attr">access-key:</span> <span class="string">xxxx</span></span><br><span class="line">    <span class="comment"># 商用版本请求超时时间</span></span><br><span class="line">    <span class="attr">timeoutMillis:</span> <span class="number">15000</span></span><br><span class="line">  <span class="comment"># 如果需要设置消费者，可以按照同样的方式集成</span></span><br><span class="line">  <span class="attr">consumer:</span></span><br><span class="line">  	<span class="attr">group:</span> <span class="string">testGroup</span></span><br><span class="line">    <span class="attr">secret-key:</span> <span class="string">xxx</span></span><br><span class="line">    <span class="attr">access-key:</span> <span class="string">xxxx</span></span><br><span class="line">    <span class="comment"># 商用版本请求超时时间 15秒</span></span><br><span class="line">    <span class="attr">timeoutMillis:</span> <span class="number">15000</span></span><br></pre></td></tr></table></figure>

<blockquote>
<p>注意这里仅仅配置了生产者，读者可以按需设置为消费者，设置方式和生产者同理。</p>
</blockquote>
<p>设置YAML之后，我们需要在Maven中引入下面的配置：</p>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>com.aliyun.openservices<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>ons-client<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">    <span class="comment">&lt;!--以下版本号请替换为Java SDK的最新版本号--&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">version</span>&gt;</span>1.8.8.1.Final<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line"><span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br></pre></td></tr></table></figure>

<p>最后应该如何使用呢？这里就是商用RocketMq比较蛋疼的点了，使用<code>RocketMQTemplate</code>是这种情况下是无法使用商用RocketMq的，我们需要 <strong>手动注入商用的SDK依赖ProducerBean</strong>，具体的操作如下：</p>
<ol>
<li>构建<strong>配置类</strong>，这里仿照了官方提供的demo增减配置：</li>
</ol>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment">阿里云服务配置封装，注意和本地部署的rocketmq配置区分</span></span><br><span class="line"><span class="comment">*/</span></span><br><span class="line"><span class="meta">@Data</span></span><br><span class="line"><span class="meta">@Configuration</span></span><br><span class="line"><span class="meta">@ConfigurationProperties</span>(prefix = <span class="string">"rocketmq"</span>)</span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">AliyunRocketMqConfig</span> </span>&#123;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     *鉴权需要的AccessKey ID</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Value</span>(<span class="string">"$&#123;rocketmq.use-aliyun-rocketSever:null&#125;"</span>)</span><br><span class="line">    <span class="keyword">private</span> String useAliyunRocketMqServerEnable;</span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     *鉴权需要的AccessKey ID</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Value</span>(<span class="string">"$&#123;rocketmq.producer.access-key:null&#125;"</span>)</span><br><span class="line">    <span class="keyword">private</span> String accessKey;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     *鉴权需要的AccessKey Secret</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Value</span>(<span class="string">"$&#123;rocketmq.producer.secret-key:null&#125;"</span>)</span><br><span class="line">    <span class="keyword">private</span> String secretKey;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 实例TCP 协议公网接入地址（实际项目，填写自己阿里云MQ的公网地址）</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Value</span>(<span class="string">"$&#123;rocketmq.name-server:null&#125;"</span>)</span><br><span class="line">    <span class="keyword">private</span> String nameSrvAddr;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 延时队列group</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Value</span>(<span class="string">"$&#123;rocketmq.producer.group:null&#125;"</span>)</span><br><span class="line">    <span class="keyword">private</span> String groupId;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 消息发送超时时间，如果服务端在配置的对应时间内未ACK，则发送客户端认为该消息发送失败。</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Value</span>(<span class="string">"$&#123;rocketmq.producer.timeoutMillis:null&#125;"</span>)</span><br><span class="line">    <span class="keyword">private</span> String timeoutMillis;</span><br><span class="line"></span><br><span class="line"></span><br><span class="line">	<span class="comment">//获取Properties</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> Properties <span class="title">getRocketMqProperty</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        Properties properties = <span class="keyword">new</span> Properties();</span><br><span class="line">        properties.setProperty(PropertyKeyConst.GROUP_ID,<span class="keyword">this</span>.getGroupId());</span><br><span class="line">        properties.setProperty(PropertyKeyConst.AccessKey, <span class="keyword">this</span>.accessKey);</span><br><span class="line">        properties.setProperty(PropertyKeyConst.SecretKey, <span class="keyword">this</span>.secretKey);</span><br><span class="line">        properties.setProperty(PropertyKeyConst.NAMESRV_ADDR, <span class="keyword">this</span>.nameSrvAddr);</span><br><span class="line">        properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, <span class="keyword">this</span>.timeoutMillis);</span><br><span class="line">        <span class="keyword">return</span> properties;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<ol start="2">
<li>构建商用RocketMq初始化类。这里会遇到<strong>比较蛋疼的事情</strong>，因为我们的依赖是商用RocketMq与开源的SpringBoot依赖共存的，虽然我们可以商用的RocketMq，但是启动的时候会执行到此类进行初始化，<strong>返回NULL会导致SpringBoot项目无法正常启动</strong>，这里无奈只能使用一个warn日志进行提示开源版本有可能出现Bean被覆盖问题，实际上使用下来没有特别大的影响。</li>
</ol>
<blockquote>
<p> 写的比较丑陋，读者有更优雅的处理方式欢迎指导。笔者目前只想到了使用这种“不管”的方式保证项目不改任何代码的情况正常运行。</p>
</blockquote>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment"> * 阿里云rocketMq初始化</span></span><br><span class="line"><span class="comment"> **/</span></span><br><span class="line"><span class="meta">@Configuration</span></span><br><span class="line"><span class="meta">@Slf</span>4j</span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">AliyunProducerInit</span> </span>&#123;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Autowired</span></span><br><span class="line">    <span class="keyword">private</span> AliyunRocketMqConfig aliyunRocketMqConfig;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Bean</span>(initMethod = <span class="string">"start"</span>, destroyMethod = <span class="string">"shutdown"</span>)</span><br><span class="line">    <span class="function"><span class="keyword">public</span> ProducerBean <span class="title">buildProducer</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        <span class="keyword">if</span>(!Boolean.valueOf(aliyunRocketMqConfig.getUseAliyunRocketMqServerEnable()))&#123;</span><br><span class="line">            log.warn(<span class="string">"非商用版本为了兼容依然需要注入此Bean，但是只读取有关nameServ和group信息"</span>);</span><br><span class="line">        &#125;</span><br><span class="line">        ProducerBean producer = <span class="keyword">new</span> ProducerBean();</span><br><span class="line">        <span class="comment">// ProducerBean中的properties只有被覆盖的配置会使用自定义配置，其他配置会使用SDK的默认配置。</span></span><br><span class="line">        producer.setProperties(aliyunRocketMqConfig.getRocketMqProperty());</span><br><span class="line">        <span class="keyword">return</span> producer;</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>



<p>此外这里必须要吐槽一下商用RocketMq的注释居然是<strong>全中文</strong>的！比如<code>com.aliyun.openservices.ons.api.bean.ProducerBean</code>的注释：</p>
<p><img src="https://adong-picture.oss-cn-shenzhen.aliyuncs.com/adong/image-20221229195508260.png" alt="producerBean"></p>
<blockquote>
<p> 这样是好事还是坏事，大家自行体会。。。。。个人第一次看到的时候着实被震惊了。</p>
</blockquote>
<ol start="3">
<li>做完上面两步之后，我们就可以实现<code>RocketMqTemplate</code>调用请求，至此完成兼容。</li>
</ol>
<h2 id="SpringBoot项目兼容小结"><a href="#SpringBoot项目兼容小结" class="headerlink" title="SpringBoot项目兼容小结"></a>SpringBoot项目兼容小结</h2><p>这里简单小结一下SpringBoot的兼容过程，可以看到整个步骤仅仅是 在商用RocketMq多做了一步bean注入的操作而已，整体使用上十分简单。但是这里只介绍了生产者的集成，那么消费者如何兼容？稍安勿躁，我们接着看Spring Cloud版本的集成案例。</p>
<h2 id="Spring-Cloud-Alibaba项目兼容"><a href="#Spring-Cloud-Alibaba项目兼容" class="headerlink" title="Spring Cloud Alibaba项目兼容"></a>Spring Cloud Alibaba项目兼容</h2><p>目前国内使用的比较多的是<strong>Spring Cloud Alibaba</strong>，注意这些配置都写入到Nacos当中。</p>
<h3 id="开源版本-1"><a href="#开源版本-1" class="headerlink" title="开源版本"></a>开源版本</h3><p>开源版本的接入方式和SpringBoot是一样的，这里简单回顾：</p>
<ol>
<li>开源版本需要设置参数，这里设置了生产者和消费者：</li>
</ol>
<figure class="highlight yaml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">#消息队列 - 开源版本</span></span><br><span class="line"><span class="attr">rocketmq:</span></span><br><span class="line">  <span class="attr">use-aliyun-rocketSever:</span> <span class="literal">false</span></span><br><span class="line">  <span class="attr">name-server:</span> <span class="number">192.168</span><span class="number">.0</span><span class="number">.92</span><span class="string">:9876</span></span><br><span class="line">  <span class="attr">producer:</span></span><br><span class="line">    <span class="attr">group:</span> <span class="string">testGroup</span></span><br><span class="line">    <span class="attr">secret-key:</span> <span class="string">NONE</span></span><br><span class="line">    <span class="attr">access-key:</span> <span class="string">NONE</span></span><br><span class="line">    <span class="attr">timeoutMillis:</span> <span class="number">15000</span></span><br><span class="line">    <span class="attr">consumeThreadNums:</span> <span class="number">20</span></span><br><span class="line">  <span class="attr">consumer:</span></span><br><span class="line">    <span class="attr">group:</span> <span class="string">testGroup</span></span><br><span class="line">    <span class="attr">secret-key:</span> <span class="string">NONE</span></span><br><span class="line">    <span class="attr">access-key:</span> <span class="string">NONE</span></span><br><span class="line">    <span class="attr">timeoutMillis:</span> <span class="number">15000</span></span><br><span class="line">    <span class="attr">consumeThreadNums:</span> <span class="number">20</span></span><br></pre></td></tr></table></figure>

<ol start="2">
<li>添加依赖：</li>
</ol>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>org.apache.rocketmq<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>rocketmq-spring-boot<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">version</span>&gt;</span>2.2.2<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line"><span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br></pre></td></tr></table></figure>

<ol start="3">
<li>使用<code>RocketMqTemplate</code>可以进行消息发送，而消费者则需要使用监听器+注解的方式，快速注入一个消费者。大体模板如下：</li>
</ol>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@Slf</span>4j</span><br><span class="line"><span class="meta">@Component</span></span><br><span class="line"><span class="meta">@RocketMQMessageListener</span>(topic = <span class="string">"test_topic"</span>, consumerGroup = <span class="string">"testGroup"</span>, selectorExpression = <span class="string">"test_tag"</span>)</span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">QueueRemoteRecoListener</span> <span class="keyword">implements</span> <span class="title">RocketMQListener</span>&lt;<span class="title">MessageExt</span>&gt; </span>&#123;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">onMessage</span><span class="params">(MessageExt message)</span> </span>&#123;</span><br><span class="line">        <span class="comment">// dosomething</span></span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>



<h3 id="商用版本-1"><a href="#商用版本-1" class="headerlink" title="商用版本"></a>商用版本</h3><p>这节是本文稍微复杂一点的部分，我们按照步骤介绍接入过程：</p>
<ol>
<li>在Nacos的配置中加入RocketMq商用所需的配置内容，和开源版本的设置类似：</li>
</ol>
<figure class="highlight yaml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">#消息队列 - 商用版本</span></span><br><span class="line"><span class="attr">rocketmq:</span></span><br><span class="line"> <span class="comment"># 开源RocketMq和商业版RocketMq切换开关</span></span><br><span class="line"><span class="attr">use-aliyun-rocketSever:</span> <span class="literal">true</span></span><br><span class="line"><span class="attr">name-server:</span> <span class="string">http://xxxx.mq.aliyuncs.com:80</span></span><br><span class="line"><span class="attr">producer:</span></span><br><span class="line"><span class="comment"># 目前uat借助开发测试使用</span></span><br><span class="line">  <span class="attr">group:</span> <span class="string">testGroup</span></span><br><span class="line">  <span class="attr">secret-key:</span> <span class="string">xxx</span></span><br><span class="line">  <span class="attr">access-key:</span> <span class="string">xxx</span></span><br><span class="line">  <span class="attr">timeoutMillis:</span> <span class="number">15000</span></span><br><span class="line">  <span class="attr">consumeThreadNums:</span> <span class="number">20</span></span><br><span class="line"><span class="attr">consumer:</span></span><br><span class="line">  <span class="attr">group:</span> <span class="string">testGroup</span></span><br><span class="line">  <span class="attr">secret-key:</span> <span class="string">xxx</span></span><br><span class="line">  <span class="attr">access-key:</span> <span class="string">xxx</span></span><br><span class="line">  <span class="attr">timeoutMillis:</span> <span class="number">15000</span></span><br><span class="line">  <span class="attr">consumeThreadNums:</span> <span class="number">20</span></span><br></pre></td></tr></table></figure>

<ol start="2">
<li>添加商用RocketMq的TCP接入方式需要的依赖包。</li>
</ol>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>com.aliyun.openservices<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>ons-client<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">    <span class="comment">&lt;!--以下版本号请替换为Java SDK的最新版本号--&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">version</span>&gt;</span>1.8.8.1.Final<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line"><span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br></pre></td></tr></table></figure>

<ol start="3">
<li>新增<strong>配置类</strong>，和SpringBoot的商用版本方式也是类似的：</li>
</ol>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br><span class="line">67</span><br><span class="line">68</span><br><span class="line">69</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment"> * </span></span><br><span class="line"><span class="comment"> * rocketmq 阿里云服务配置封装，注意和本地部署的rocketmq配置区分</span></span><br><span class="line"><span class="comment"> **/</span></span><br><span class="line"><span class="meta">@Data</span></span><br><span class="line"><span class="meta">@Configuration</span></span><br><span class="line"><span class="meta">@ConfigurationProperties</span>(prefix = <span class="string">"rocketmq"</span>)</span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">AliyunCommercialRocketMqConfig</span> </span>&#123;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     *鉴权需要的AccessKey ID</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Value</span>(<span class="string">"$&#123;rocketmq.use-aliyun-rocketSever:null&#125;"</span>)</span><br><span class="line">    <span class="keyword">private</span> String useAliyunRocketMqServerEnable;</span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     *鉴权需要的AccessKey ID</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Value</span>(<span class="string">"$&#123;rocketmq.consumer.access-key:null&#125;"</span>)</span><br><span class="line">    <span class="keyword">private</span> String accessKey;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     *</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Value</span>(<span class="string">"$&#123;rocketmq.use-aliyun-rocketsever:null&#125;"</span>)</span><br><span class="line">    <span class="keyword">private</span> String useAliyunRocketServer;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     *鉴权需要的AccessKey Secret</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Value</span>(<span class="string">"$&#123;rocketmq.consumer.secret-key:null&#125;"</span>)</span><br><span class="line">    <span class="keyword">private</span> String secretKey;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 实例TCP 协议公网接入地址（实际项目，填写自己阿里云MQ的公网地址）</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Value</span>(<span class="string">"$&#123;rocketmq.name-server:null&#125;"</span>)</span><br><span class="line">    <span class="keyword">private</span> String nameSrvAddr;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 延时队列group</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Value</span>(<span class="string">"$&#123;rocketmq.consumer.group:null&#125;"</span>)</span><br><span class="line">    <span class="keyword">private</span> String groupId;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 消息发送超时时间，如果服务端在配置的对应时间内未ACK，则发送客户端认为该消息发送失败。</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Value</span>(<span class="string">"$&#123;rocketmq.consumer.timeoutMillis:null&#125;"</span>)</span><br><span class="line">    <span class="keyword">private</span> String timeoutMillis;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 将消费者线程数固定为20个 20为默认值</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Value</span>(<span class="string">"$&#123;rocketmq.consumer.consumeThreadNums:null&#125;"</span>)</span><br><span class="line">    <span class="keyword">private</span> String consumeThreadNums;</span><br><span class="line"></span><br><span class="line"></span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> Properties <span class="title">getRocketMqProperty</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        Properties properties = <span class="keyword">new</span> Properties();</span><br><span class="line">        properties.setProperty(PropertyKeyConst.GROUP_ID,<span class="keyword">this</span>.getGroupId());</span><br><span class="line">        properties.setProperty(PropertyKeyConst.AccessKey, <span class="keyword">this</span>.accessKey);</span><br><span class="line">        properties.setProperty(PropertyKeyConst.SecretKey, <span class="keyword">this</span>.secretKey);</span><br><span class="line">        properties.setProperty(PropertyKeyConst.NAMESRV_ADDR, <span class="keyword">this</span>.nameSrvAddr);</span><br><span class="line">        properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, <span class="keyword">this</span>.timeoutMillis);</span><br><span class="line">        <span class="keyword">return</span> properties;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<ol start="4">
<li>下一步是扩展官方<strong>消息订阅类</strong>，为啥要这样做？主要是接入实验中发现官方的消息订阅类没有 <strong>全量参数</strong>的构造器，难以对于消息订阅类<strong>静态参数化</strong>常量，构造一个订阅消息使用默认方式还需要static代码块设置。所以这里对于官方的消息订阅类进行扩展，子类代码是对于父类拷贝了一遍，没有做其他改动：</li>
</ol>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment">对于指定类进行扩展，更容易的初始化</span></span><br><span class="line"><span class="comment"><span class="doctag">@see</span> com.aliyun.openservices.ons.api.bean.Subscription 被扩展类</span></span><br><span class="line"><span class="comment"> **/</span></span><br><span class="line"><span class="meta">@Data</span></span><br><span class="line"><span class="meta">@ToString</span></span><br><span class="line"><span class="meta">@AllArgsConstructor</span></span><br><span class="line"><span class="meta">@NoArgsConstructor</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">AliyunCommercialRocketMqSubscriptionExt</span> <span class="keyword">extends</span> <span class="title">Subscription</span> </span>&#123;</span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 主题</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="keyword">private</span> String topic;</span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 条件表达式，具体参考rocketmq</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="keyword">private</span> String expression;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * TAG or SQL92</span></span><br><span class="line"><span class="comment">     * &lt;br&gt;if null, equals to TAG</span></span><br><span class="line"><span class="comment">     *</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@see</span> com.aliyun.openservices.ons.api.ExpressionType#TAG</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@see</span> com.aliyun.openservices.ons.api.ExpressionType#SQL92</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="keyword">private</span> String type;</span><br><span class="line"></span><br><span class="line"></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">int</span> <span class="title">hashCode</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        <span class="keyword">final</span> <span class="keyword">int</span> prime = <span class="number">31</span>;</span><br><span class="line">        <span class="keyword">int</span> result = <span class="number">1</span>;</span><br><span class="line">        result = prime * result + ((topic == <span class="keyword">null</span>) ? <span class="number">0</span> : topic.hashCode());</span><br><span class="line">        <span class="keyword">return</span> result;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">boolean</span> <span class="title">equals</span><span class="params">(Object obj)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">if</span> (<span class="keyword">this</span> == obj) &#123;</span><br><span class="line">            <span class="keyword">return</span> <span class="keyword">true</span>;</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="keyword">if</span> (obj == <span class="keyword">null</span>) &#123;</span><br><span class="line">            <span class="keyword">return</span> <span class="keyword">false</span>;</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="keyword">if</span> (getClass() != obj.getClass()) &#123;</span><br><span class="line">            <span class="keyword">return</span> <span class="keyword">false</span>;</span><br><span class="line">        &#125;</span><br><span class="line">        AliyunCommercialRocketMqSubscriptionExt other = (AliyunCommercialRocketMqSubscriptionExt) obj;</span><br><span class="line">        <span class="keyword">if</span> (topic == <span class="keyword">null</span>) &#123;</span><br><span class="line">            <span class="keyword">if</span> (other.topic != <span class="keyword">null</span>) &#123;</span><br><span class="line">                <span class="keyword">return</span> <span class="keyword">false</span>;</span><br><span class="line">            &#125;</span><br><span class="line">        &#125; <span class="keyword">else</span> <span class="keyword">if</span> (!topic.equals(other.topic)) &#123;</span><br><span class="line">            <span class="keyword">return</span> <span class="keyword">false</span>;</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="keyword">return</span> <span class="keyword">true</span>;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<ol start="5">
<li>接着我们构建静态化的消息订阅类，这个常量类会封装系统需要使用到的消息订阅对象，在后续注册监听器需要使用，这里先提前定义。</li>
</ol>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment"> * 队列常规配置，用于启动时候初始化配置，</span></span><br><span class="line"><span class="comment"> * 所有配置均需要放置到此类中对外扩展使用</span></span><br><span class="line"><span class="comment"> **/</span></span><br><span class="line"><span class="keyword">public</span> <span class="keyword">final</span> <span class="class"><span class="keyword">class</span> <span class="title">AliyunCommercialRocketMqConstants</span> </span>&#123;</span><br><span class="line">    </span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">final</span> String TEST_TOPIC = <span class="string">"test_topic"</span>;</span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">final</span> String TEST_TAG = <span class="string">"test_tag"</span>;</span><br><span class="line">    </span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 参考案例，仅仅作为开发验证使用</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">final</span> AliyunCommercialRocketMqSubscriptionExt QUEUE_TEST = <span class="keyword">new</span> AliyunCommercialRocketMqSubscriptionExt(RocketMqKey.TEST_TOPIC, RocketMqKey.TEST_TAG, <span class="keyword">null</span>);</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<p>为了方便管理，我们可以把这些关键Key在单独放到一个类：</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">RocketMqKey</span> </span>&#123;</span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">final</span> String TEST_TOPIC = <span class="string">"test_topic"</span>;</span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">final</span> String TEST_TAG = <span class="string">"test_tag"</span>;   </span><br><span class="line">&#125;</span><br><span class="line"></span><br><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment"> * 队列常规配置，用于启动时候初始化配置，</span></span><br><span class="line"><span class="comment"> * 所有配置均需要放置到此类中对外扩展使用</span></span><br><span class="line"><span class="comment"> **/</span></span><br><span class="line"><span class="keyword">public</span> <span class="keyword">final</span> <span class="class"><span class="keyword">class</span> <span class="title">AliyunCommercialRocketMqConstants</span> </span>&#123;</span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 参考案例，仅仅作为开发验证使用</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">final</span> AliyunCommercialRocketMqSubscriptionExt QUEUE_TEST = <span class="keyword">new</span> AliyunCommercialRocketMqSubscriptionExt(RocketMqKey.TEST_TOPIC, RocketMqKey.TEST_TAG, <span class="keyword">null</span>);</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<ol start="6">
<li>做好一系列准备之后，我们开始<strong>商用版本生产者兼容</strong>，商用版本的发送者可以像是下面这样封装官方提供的demo，也可以使用<strong>注入ProducerBean</strong>的方式集成：</li>
</ol>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br><span class="line">67</span><br><span class="line">68</span><br><span class="line">69</span><br><span class="line">70</span><br><span class="line">71</span><br><span class="line">72</span><br><span class="line">73</span><br><span class="line">74</span><br><span class="line">75</span><br><span class="line">76</span><br><span class="line">77</span><br><span class="line">78</span><br><span class="line">79</span><br><span class="line">80</span><br><span class="line">81</span><br><span class="line">82</span><br><span class="line">83</span><br><span class="line">84</span><br><span class="line">85</span><br><span class="line">86</span><br><span class="line">87</span><br><span class="line">88</span><br><span class="line">89</span><br><span class="line">90</span><br><span class="line">91</span><br><span class="line">92</span><br><span class="line">93</span><br><span class="line">94</span><br><span class="line">95</span><br><span class="line">96</span><br><span class="line">97</span><br><span class="line">98</span><br><span class="line">99</span><br><span class="line">100</span><br><span class="line">101</span><br><span class="line">102</span><br><span class="line">103</span><br><span class="line">104</span><br><span class="line">105</span><br><span class="line">106</span><br><span class="line">107</span><br><span class="line">108</span><br><span class="line">109</span><br><span class="line">110</span><br><span class="line">111</span><br><span class="line">112</span><br><span class="line">113</span><br><span class="line">114</span><br><span class="line">115</span><br><span class="line">116</span><br><span class="line">117</span><br><span class="line">118</span><br><span class="line">119</span><br><span class="line">120</span><br><span class="line">121</span><br><span class="line">122</span><br><span class="line">123</span><br><span class="line">124</span><br><span class="line">125</span><br><span class="line">126</span><br><span class="line">127</span><br><span class="line">128</span><br><span class="line">129</span><br><span class="line">130</span><br><span class="line">131</span><br><span class="line">132</span><br><span class="line">133</span><br><span class="line">134</span><br><span class="line">135</span><br><span class="line">136</span><br><span class="line">137</span><br><span class="line">138</span><br><span class="line">139</span><br><span class="line">140</span><br><span class="line">141</span><br><span class="line">142</span><br><span class="line">143</span><br><span class="line">144</span><br><span class="line">145</span><br><span class="line">146</span><br><span class="line">147</span><br><span class="line">148</span><br><span class="line">149</span><br><span class="line">150</span><br><span class="line">151</span><br><span class="line">152</span><br><span class="line">153</span><br><span class="line">154</span><br><span class="line">155</span><br><span class="line">156</span><br><span class="line">157</span><br><span class="line">158</span><br><span class="line">159</span><br><span class="line">160</span><br><span class="line">161</span><br><span class="line">162</span><br><span class="line">163</span><br><span class="line">164</span><br><span class="line">165</span><br><span class="line">166</span><br><span class="line">167</span><br><span class="line">168</span><br><span class="line">169</span><br><span class="line">170</span><br><span class="line">171</span><br><span class="line">172</span><br><span class="line">173</span><br><span class="line">174</span><br><span class="line">175</span><br><span class="line">176</span><br><span class="line">177</span><br><span class="line">178</span><br><span class="line">179</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment"> * 商用aliyun Rocketmq 工具类封装</span></span><br><span class="line"><span class="comment"> **/</span></span><br><span class="line"><span class="meta">@Component</span></span><br><span class="line"><span class="meta">@Slf</span>4j</span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">AliyunCommercialRocketMqSendUtils</span> </span>&#123;</span><br><span class="line"></span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">static</span> <span class="keyword">final</span> <span class="keyword">long</span> KEEP_ALIVE_TIME = <span class="number">60L</span>;</span><br><span class="line"></span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">final</span> ThreadPoolExecutor THREAD_POOL_EXECUTOR = <span class="keyword">new</span> ThreadPoolExecutor(<span class="number">0</span>, Integer.MAX_VALUE,</span><br><span class="line">            KEEP_ALIVE_TIME, TimeUnit.SECONDS,</span><br><span class="line">            <span class="keyword">new</span> SynchronousQueue&lt;&gt;());</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Autowired</span></span><br><span class="line">    <span class="keyword">private</span> Producer producer;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Autowired</span></span><br><span class="line">    <span class="keyword">private</span> SaMqMessageFailService saMqMessageFailService;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Autowired</span></span><br><span class="line">    <span class="keyword">private</span> AliyunRocketMqConfig aliyunRocketMqConfig;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 异步推送，需要调用方手动指定callback</span></span><br><span class="line"><span class="comment">     </span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">singleAsyncSend</span><span class="params">(String topic, String tag, String msgBody, SendCallback sendCallback)</span> </span>&#123;</span><br><span class="line">        singleAsyncSend(topic, tag, msgBody, <span class="keyword">null</span>, sendCallback);</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 异步推送，需要调用方手动指定callback</span></span><br><span class="line"><span class="comment">     * 如果需要使用callback通知，可以使用下面的代码进行处理</span></span><br><span class="line"><span class="comment">     &lt;pre&gt;</span></span><br><span class="line"><span class="comment">         producer.sendAsync(msg, new SendCallback() &#123;</span></span><br><span class="line"><span class="comment">        <span class="doctag">@Override</span></span></span><br><span class="line"><span class="comment">        public void onSuccess(final SendResult sendResult) &#123;</span></span><br><span class="line"><span class="comment">            assert sendResult != null;</span></span><br><span class="line"><span class="comment">            System.out.println(sendResult);</span></span><br><span class="line"><span class="comment">        &#125;</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment">        <span class="doctag">@Override</span></span></span><br><span class="line"><span class="comment">        public void onException(final OnExceptionContext context) &#123;</span></span><br><span class="line"><span class="comment">            ONSClientException exception = context.getException();</span></span><br><span class="line"><span class="comment">            //                    //出现异常意味着发送失败，为了避免消息丢失，建议缓存该消息然后进行重试。</span></span><br><span class="line"><span class="comment">            log.error("【RocketMq-Commercial】发送失败，消息存储到失败记录表，发送 topic =&gt; &#123;&#125;, 发送 tag =&gt; &#123;&#125;， msgBody =&gt; &#123;&#125;， key =&gt; &#123;&#125;（如果设置key，可以使用key查找消息）,商用版需要在RocketMq云服务重试，失败原因为: &#123;&#125;"</span></span><br><span class="line"><span class="comment">            , topic, tag, msgBody, key, exception.getMessage());</span></span><br><span class="line"><span class="comment">            buildMqErrorInfoAndInsertDb(msgBody, exception);</span></span><br><span class="line"><span class="comment">        &#125;</span></span><br><span class="line"><span class="comment">    &#125;);</span></span><br><span class="line"><span class="comment">     &lt;/pre&gt;</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@param</span> topic 主题</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@return</span> void</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@description</span></span></span><br><span class="line"><span class="comment">     * <span class="doctag">@param</span>: tag  标签</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@param</span>: msgBody 推送消息</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@param</span>: key 方便消息查找的key</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@param</span>: sendCallback 手动回调</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">singleAsyncSend</span><span class="params">(String topic, String tag, String msgBody, String key, SendCallback sendCallback)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">if</span>(Boolean.FALSE.equals(Boolean.valueOf(aliyunRocketMqConfig.getUseAliyunRocketMqServerEnable())))&#123;</span><br><span class="line">            <span class="keyword">throw</span> <span class="keyword">new</span> UnsupportedOperationException(<span class="string">"请设置 UseAliyunRocketServer=true 切换到商用rocketMq模式"</span>);</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="keyword">if</span> (StringUtils.isAnyBlank(topic, tag, msgBody)) &#123;</span><br><span class="line">            <span class="keyword">throw</span> <span class="keyword">new</span> IllegalArgumentException(<span class="string">"topic, tag and msgBody must be not blank"</span>);</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="comment">//对于使用异步接口，建议设置单独的回调处理线程池，拥有更灵活的配置和监控能力。</span></span><br><span class="line">        <span class="comment">//如下构造线程的方式请求队列为无界仅用作示例，有OOM的风险。</span></span><br><span class="line">        <span class="comment">//更合理的构造方式请参考阿里巴巴Java开发手册： https://github.com/alibaba/p3c</span></span><br><span class="line">        producer.setCallbackExecutor(THREAD_POOL_EXECUTOR);</span><br><span class="line"></span><br><span class="line">        <span class="comment">//循环发送消息</span></span><br><span class="line">        Message msg = <span class="keyword">new</span> Message( <span class="comment">//</span></span><br><span class="line">                <span class="comment">// Message所属的Topic</span></span><br><span class="line">                topic,</span><br><span class="line">                <span class="comment">// Message Tag 可理解为Gmail中的标签，对消息进行再归类，方便Consumer指定过滤条件在MQ服务器过滤</span></span><br><span class="line">                tag,</span><br><span class="line">                <span class="comment">// Message Body 可以是任何二进制形式的数据， MQ不做任何干预</span></span><br><span class="line">                <span class="comment">// 需要Producer与Consumer协商好一致的序列化和反序列化方式</span></span><br><span class="line">                msgBody.getBytes());</span><br><span class="line">        <span class="comment">// 设置代表消息的业务关键属性，请尽可能全局唯一</span></span><br><span class="line">        <span class="comment">// 以方便您在无法正常收到消息情况下，可通过MQ 控制台查询消息并补发</span></span><br><span class="line">        <span class="comment">// 注意：不设置也不会影响消息正常收发</span></span><br><span class="line">        <span class="keyword">if</span> (CharSequenceUtil.isNotBlank(key)) &#123;</span><br><span class="line">            msg.setKey(key);</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="comment">// 发送消息，只要不抛异常就是成功</span></span><br><span class="line">        <span class="keyword">try</span> &#123;</span><br><span class="line">            producer.sendAsync(msg, sendCallback);</span><br><span class="line">        &#125; <span class="keyword">catch</span> (ONSClientException exception) &#123;</span><br><span class="line">            <span class="comment">//出现异常意味着发送失败，为了避免消息丢失，建议缓存该消息然后进行重试。</span></span><br><span class="line">            log.error(<span class="string">"【RocketMq-Commercial】发送失败，消息存储到失败记录表，发送 topic =&gt; &#123;&#125;, 发送 tag =&gt; &#123;&#125;， msgBody =&gt; &#123;&#125;， key =&gt; &#123;&#125;（如果设置key，可以使用key查找消息）,商用版需要在RocketMq云服务重试，失败原因为: &#123;&#125;"</span></span><br><span class="line">                    , topic, tag, msgBody, key, exception.getMessage());</span><br><span class="line">            buildMqErrorInfoAndInsertDb(msgBody, exception);</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line"></span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 阻塞单独推送队列，使用系统默认的Key生成规则</span></span><br><span class="line"><span class="comment">     *</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@param</span> topic 主题</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@return</span> void</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@description</span> 阻塞单独推送队列</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@param</span>: tag 标签</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@param</span>: msgBody msgBody内容</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">singleSyncSend</span><span class="params">(String topic, String tag, String msgBody)</span> </span>&#123;</span><br><span class="line">        singleSyncSend(topic, tag, msgBody, <span class="keyword">null</span>);</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 阻塞单独推送队列，使用系统默认的Key生成规则</span></span><br><span class="line"><span class="comment">     *</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@param</span> topic 主题</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@return</span> void</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@description</span> 阻塞单独推送队列</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@param</span>: tag 标签</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@param</span>: msgBody msgBody内容</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">singleSyncSend</span><span class="params">(String topic, String tag, String msgBody, String key)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">if</span>(!Boolean.valueOf(aliyunRocketMqConfig.getUseAliyunRocketMqServerEnable()))&#123;</span><br><span class="line">            <span class="keyword">throw</span> <span class="keyword">new</span> UnsupportedOperationException(<span class="string">"请设置 UseAliyunRocketServer=true 切换到商用rocketMq模式"</span>);</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="keyword">if</span> (StringUtils.isAnyBlank(topic, tag, msgBody)) &#123;</span><br><span class="line">            <span class="keyword">throw</span> <span class="keyword">new</span> IllegalArgumentException(<span class="string">"topic, tag and msgBody must be not blank"</span>);</span><br><span class="line">        &#125;</span><br><span class="line">        Message msg = <span class="keyword">new</span> Message( <span class="comment">//</span></span><br><span class="line">                <span class="comment">// Message所属的Topic</span></span><br><span class="line">                topic,</span><br><span class="line">                <span class="comment">// Message Tag 可理解为Gmail中的标签，对消息进行再归类，方便Consumer指定过滤条件在MQ服务器过滤</span></span><br><span class="line">                tag,</span><br><span class="line">                <span class="comment">// Message Body 可以是任何二进制形式的数据， MQ不做任何干预</span></span><br><span class="line">                <span class="comment">// 需要Producer与Consumer协商好一致的序列化和反序列化方式</span></span><br><span class="line">                msgBody.getBytes());</span><br><span class="line">        <span class="comment">// 设置代表消息的业务关键属性，请尽可能全局唯一</span></span><br><span class="line">        <span class="comment">// 以方便您在无法正常收到消息情况下，可通过MQ 控制台查询消息并补发</span></span><br><span class="line">        <span class="comment">// 注意：不设置也不会影响消息正常收发</span></span><br><span class="line">        <span class="keyword">if</span> (CharSequenceUtil.isNotBlank(key)) &#123;</span><br><span class="line">            msg.setKey(key);</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="comment">// 发送消息，只要不抛异常就是成功</span></span><br><span class="line">        <span class="keyword">try</span> &#123;</span><br><span class="line">            SendResult sendResult = producer.send(msg);</span><br><span class="line">            log.info(<span class="string">"【RocketMq-Commercial】推送成功，singleSyncSend =&gt; &#123;&#125;"</span>, JSON.toJSONString(sendResult));</span><br><span class="line">        &#125; <span class="keyword">catch</span> (ONSClientException e) &#123;</span><br><span class="line">            log.error(<span class="string">"【RocketMq-Commercial】发送失败，消息存储到失败记录表，发送 topic =&gt; &#123;&#125;, 发送 tag =&gt; &#123;&#125;， msgBody =&gt; &#123;&#125;， key =&gt; &#123;&#125;（如果设置key，可以使用key查找消息）,商用版需要在RocketMq云服务重试，失败原因为: &#123;&#125;"</span></span><br><span class="line">                    , topic, tag, msgBody, key, e.getMessage());</span><br><span class="line">            buildMqErrorInfoAndInsertDb(msgBody, e);</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 构建mq错误信息，并且插入到异常推送表</span></span><br><span class="line"><span class="comment">     *</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@param</span> msgBody</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@return</span> void</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@description</span> 构建mq错误信息，并且插入到异常推送表</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@param</span>: e</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="function"><span class="keyword">private</span> <span class="keyword">void</span> <span class="title">buildMqErrorInfoAndInsertDb</span><span class="params">(String msgBody, ONSClientException e)</span> </span>&#123;</span><br><span class="line">        <span class="comment">//出现异常意味着发送失败，为了避免消息丢失，建议缓存该消息然后进行重试。</span></span><br><span class="line">        <span class="comment">//出现异常，保存异常信息到数据库</span></span><br><span class="line">        SaMqMessageFail saMqMessageFail = <span class="keyword">new</span> SaMqMessageFail();</span><br><span class="line">        saMqMessageFail.setMqId(IdUtil.randomUUID());</span><br><span class="line">        saMqMessageFail.setQueueName(<span class="string">"RocketMq-Commercial"</span>);</span><br><span class="line">        saMqMessageFail.setQueueMessage(msgBody);</span><br><span class="line">        <span class="comment">// 2 代表失败</span></span><br><span class="line">        saMqMessageFail.setStatus(<span class="number">2</span>);</span><br><span class="line">        <span class="comment">// 1 代表重试次数</span></span><br><span class="line">        saMqMessageFail.setProcCount(<span class="number">1</span>);</span><br><span class="line">        saMqMessageFail.setFailReason(e.toString());</span><br><span class="line">        saMqMessageFail.setCreateDate(<span class="keyword">new</span> Date());</span><br><span class="line">        saMqMessageFailService.insert(saMqMessageFail);</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<blockquote>
<p>注入ProductBean的方式可以从上面提到的SpringBoot集成方式处理。</p>
</blockquote>
<ol start="7">
<li>现在我们解决之前遗留的问题，如果是消费者，应该如何更优雅的接收消息？首先我们需要明白，商用RocketMq注入是需要依靠手动构建<strong>监听器</strong>，但是我们上面提到SpringBoot提供了<code>注解+ @Component</code>的方式实现队列监听的消费者。此外为了避免和开源版本冲突，我们使用之前参数配置自定义的开关，在遇到开源版本的时候我们<strong>返回null</strong>（这里返回null Spring会扫描获取SpringBoot的RocketMq依赖，不会出现报错和无法启动的问题，和前面的情况略有不同）防止自动注入商用RocketMq的监听器。</li>
</ol>
<p>下面是商用版本注入ConsumerBean的代码，订阅商用RocketMq的消费监听器：</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment"> *  阿里云商用Rocketmq 队列接收</span></span><br><span class="line"><span class="comment"> *  商用rocketmqConsumer的所有请求会进入一个入口，需要分发到不同的具体业务处理。</span></span><br><span class="line"><span class="comment"> **/</span></span><br><span class="line"><span class="meta">@Component</span></span><br><span class="line"><span class="meta">@Slf</span>4j</span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">AliyunCommercialRocketMqQueueConsumer</span> </span>&#123;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Autowired</span></span><br><span class="line">    <span class="keyword">private</span> AliyunCommercialRocketMqConfig aliyunCommercialRocketMqConfig;</span><br><span class="line"></span><br><span class="line">    <span class="comment">// 自定义的监听器</span></span><br><span class="line">    <span class="meta">@Autowired</span></span><br><span class="line">    <span class="keyword">private</span> AliyunCommerciaRocketMqTestListener aliyunCommerciaRocketMqTestListener;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Bean</span>(initMethod = <span class="string">"start"</span>, destroyMethod = <span class="string">"shutdown"</span>)</span><br><span class="line">    <span class="function"><span class="keyword">public</span> ConsumerBean <span class="title">buildConsumer</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        <span class="comment">// 如果开关为false, 则不能注入此对象，否则商用的API会顶替掉 框架诸如的Bean出现异常</span></span><br><span class="line">        <span class="keyword">if</span>(!Boolean.valueOf(aliyunCommercialRocketMqConfig.getUseAliyunRocketServer()))&#123;</span><br><span class="line">            log.warn(<span class="string">"非商用版本不注入商用版本监听器"</span>);</span><br><span class="line">            <span class="keyword">return</span> <span class="keyword">null</span>;</span><br><span class="line">        &#125;</span><br><span class="line">        ConsumerBean consumerBean = <span class="keyword">new</span> ConsumerBean();</span><br><span class="line">        <span class="comment">//配置文件</span></span><br><span class="line">        Properties properties = aliyunCommercialRocketMqConfig.getRocketMqProperty();</span><br><span class="line">        properties.setProperty(PropertyKeyConst.GROUP_ID, aliyunCommercialRocketMqConfig.getGroupId());</span><br><span class="line">        <span class="comment">//将消费者线程数固定为20个 20为默认值</span></span><br><span class="line">        properties.setProperty(PropertyKeyConst.ConsumeThreadNums, aliyunCommercialRocketMqConfig.getConsumeThreadNums());</span><br><span class="line">        consumerBean.setProperties(properties);</span><br><span class="line">        <span class="comment">//订阅关系</span></span><br><span class="line">        Map&lt;Subscription, MessageListener&gt; subscriptionTable = <span class="keyword">new</span> HashMap&lt;&gt;(<span class="number">2</span>);</span><br><span class="line">        <span class="comment">// 使用了之前定义的测试用的监听器</span></span><br><span class="line">        subscriptionTable.put(AliyunCommercialRocketMqConstants.QUEUE_TEST, aliyunCommerciaRocketMqTestListener);</span><br><span class="line">        <span class="comment">//订阅多个topic如上面设置</span></span><br><span class="line"></span><br><span class="line">        consumerBean.setSubscriptionTable(subscriptionTable);</span><br><span class="line">        <span class="keyword">return</span> consumerBean;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line"></span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<p><code>AliyunCommerciaRocketMqTestListener</code> 自定义监听器为了兼容商用和开源版本做了下面的改变，为了方便理解这里加入了相关注解：</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment"> * 监听器开发模板</span></span><br><span class="line"><span class="comment"> **/</span></span><br><span class="line"><span class="meta">@Slf</span>4j</span><br><span class="line"><span class="meta">@Component</span></span><br><span class="line"><span class="comment">// 开源版本可以兼容此注解</span></span><br><span class="line"><span class="meta">@RocketMQMessageListener</span>(topic = RocketMqKey.TEST_TOPIC, consumerGroup = RocketMqKey.TEST_GROUP, selectorExpression = RocketMqKey.TEST_TAG)</span><br><span class="line"><span class="comment">// RocketMQListener&lt;MessageExt&gt; 属于SpringBoot RocketMq的依赖</span></span><br><span class="line"><span class="comment">// MessageListener 属于商用RocketMq的SDK</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">QueueRemoteReconListener</span> <span class="keyword">implements</span> <span class="title">RocketMQListener</span>&lt;<span class="title">MessageExt</span>&gt;, <span class="title">MessageListener</span> </span>&#123;</span><br><span class="line"></span><br><span class="line"></span><br><span class="line">    <span class="comment">// 开源版本的使用</span></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">onMessage</span><span class="params">(MessageExt message)</span> </span>&#123;</span><br><span class="line">        <span class="comment">// 开源版本的SpringBoot方式，和上文介绍相同</span></span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/** </span></span><br><span class="line"><span class="comment">     * <span class="doctag">@description</span>  </span></span><br><span class="line"><span class="comment">     * <span class="doctag">@param</span> message 消息类. 一条消息由主题, 消息体以及可选的消息标签, 自定义附属键值对构成.</span></span><br><span class="line"><span class="comment">     * 注意: 我们对每条消息的自定义键值对的长度没有限制, 但所有的自定义键值对, 系统键值对序列化后, 所占空间不能超过32767字节.</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@param</span>: context 每次消费消息的上下文，供将来扩展使用</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@return</span> com.aliyun.openservices.ons.api.Action </span></span><br><span class="line"><span class="comment">     */</span> </span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> Action <span class="title">consume</span><span class="params">(Message message, ConsumeContext context)</span> </span>&#123;</span><br><span class="line">        <span class="comment">// 商用版本要求实现这个方法,Action如下，默认失败重试16次，返回ReconsumeLater会触发重试机制，所以会存在重复消费的问题</span></span><br><span class="line">		<span class="comment">//public enum Action &#123;</span></span><br><span class="line">            <span class="comment">/**</span></span><br><span class="line"><span class="comment">             * 消费成功，继续消费下一条消息</span></span><br><span class="line"><span class="comment">             */</span></span><br><span class="line">            <span class="comment">//CommitMessage,</span></span><br><span class="line">            <span class="comment">/**</span></span><br><span class="line"><span class="comment">             * 消费失败，告知服务器稍后再投递这条消息，继续消费其他消息</span></span><br><span class="line"><span class="comment">             */</span></span><br><span class="line">            <span class="comment">//ReconsumeLater,</span></span><br><span class="line">		<span class="comment">//&#125;</span></span><br><span class="line"></span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<p>以上就是消费者的兼容处理，既可以满足开源版本的注解开发要求，也可以不膨胀类的情况下沿用扩展。</p>
<h3 id="Spring-Cloud-Alibaba项目兼容小结"><a href="#Spring-Cloud-Alibaba项目兼容小结" class="headerlink" title="Spring Cloud Alibaba项目兼容小结"></a>Spring Cloud Alibaba项目兼容小结</h3><p>从个人的角度来看，在生产者的兼容上比较容易实现，按照官方的demo构建带商用RocketMq的ProducerBean即可，而消费者的集成则要复杂一些，这里为了考虑不重复的设置或者写重复代码，把关键的配置静态化，同时为了官方写的粗糙的消息订阅类做了继承覆盖的操作兼容。此外为了防止开源版本的消费者Bean被商用的监听器覆盖导致失效，使用了简单的开关来进行Bean的注入控制，写的比较粗糙，读者有更好的写法欢迎讨论。</p>
<p>总体来看来说商用版本的RocketMq集成起来略微麻烦，但是还是可以接受。此外也可以看到SDK的兼容性实际上是比较简陋的，很多地方的很像但是又完全是自己重新设计的，感觉就是一个新的团队给老团队做出来的东西做兼容的感觉，不过不就是桥接嘛，我也会，所以最后个人对付消费者兼容就用来多接口实现的兼容写法了。</p>
<p>最终的集成效果是开源版本关闭注入bean的开关，配置为了照顾Properties对于不需要的配置进行占位处理，而商用版本则打开开关，通过自定义注入监听器的方式顶替掉SpringBoot的依赖。</p>
<p>最后的效果是只需要修改RocketMq的配置，启动之后自动连接到相关的RocketMq。</p>
<p>开源版本：</p>
<figure class="highlight yaml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">#消息队列 - 开源版本</span></span><br><span class="line"><span class="attr">rocketmq:</span></span><br><span class="line">  <span class="attr">use-aliyun-rocketSever:</span> <span class="literal">false</span></span><br><span class="line">  <span class="attr">name-server:</span> <span class="number">192.168</span><span class="number">.0</span><span class="number">.92</span><span class="string">:9876</span></span><br><span class="line">  <span class="attr">producer:</span></span><br><span class="line">    <span class="attr">group:</span> <span class="string">testGroup</span></span><br><span class="line">    <span class="attr">secret-key:</span> <span class="string">NONE</span></span><br><span class="line">    <span class="attr">access-key:</span> <span class="string">NONE</span></span><br><span class="line">    <span class="attr">timeoutMillis:</span> <span class="number">15000</span></span><br><span class="line">    <span class="attr">consumeThreadNums:</span> <span class="number">20</span></span><br><span class="line">  <span class="attr">consumer:</span></span><br><span class="line">    <span class="attr">group:</span> <span class="string">testGroup</span></span><br><span class="line">    <span class="attr">secret-key:</span> <span class="string">NONE</span></span><br><span class="line">    <span class="attr">access-key:</span> <span class="string">NONE</span></span><br><span class="line">    <span class="attr">timeoutMillis:</span> <span class="number">15000</span></span><br><span class="line">    <span class="attr">consumeThreadNums:</span> <span class="number">20</span></span><br></pre></td></tr></table></figure>

<p>商用版本</p>
<figure class="highlight yaml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">#消息队列 - 商用版本</span></span><br><span class="line"><span class="attr">rocketmq:</span></span><br><span class="line"> <span class="comment"># 开源RocketMq和商业版RocketMq切换开关</span></span><br><span class="line">  <span class="attr">use-aliyun-rocketSever:</span> <span class="literal">true</span></span><br><span class="line">  <span class="attr">name-server:</span> <span class="string">http://xxxx.mq.aliyuncs.com:80</span></span><br><span class="line">  <span class="attr">producer:</span></span><br><span class="line"><span class="comment"># 目前uat借助开发测试使用</span></span><br><span class="line">    <span class="attr">group:</span> <span class="string">testGroup</span></span><br><span class="line">      <span class="attr">secret-key:</span> <span class="string">xxx</span></span><br><span class="line">      <span class="attr">access-key:</span> <span class="string">xxx</span></span><br><span class="line">      <span class="attr">timeoutMillis:</span> <span class="number">15000</span></span><br><span class="line">      <span class="attr">consumeThreadNums:</span> <span class="number">20</span></span><br><span class="line">    <span class="attr">consumer:</span></span><br><span class="line">      <span class="attr">group:</span> <span class="string">testGroup</span></span><br><span class="line">      <span class="attr">secret-key:</span> <span class="string">xxx</span></span><br><span class="line">      <span class="attr">access-key:</span> <span class="string">xxx</span></span><br><span class="line">      <span class="attr">timeoutMillis:</span> <span class="number">15000</span></span><br><span class="line">      <span class="attr">consumeThreadNums:</span> <span class="number">20</span></span><br></pre></td></tr></table></figure>



<h1 id="阿里云商用RocketMq简单介绍（4-X版本）"><a href="#阿里云商用RocketMq简单介绍（4-X版本）" class="headerlink" title="阿里云商用RocketMq简单介绍（4.X版本）"></a>阿里云商用RocketMq简单介绍（4.X版本）</h1><p>官方介绍：<span class="exturl" data-url="aHR0cHM6Ly9oZWxwLmFsaXl1bi5jb20vcHJvZHVjdC8yOTUzMC5odG1s" title="https://help.aliyun.com/product/29530.html">https://help.aliyun.com/product/29530.html<i class="fa fa-external-link"></i></span></p>
<p>有关RocketMq本身的介绍部分可以阅读：[[【RocketMq】RocketMq 扫盲]]，这里挑了商用版本个人认为需要关注的几个点进行介绍。注意这里使用的商用RocketMq版本为4.X的版本。</p>
<h2 id="计费模式"><a href="#计费模式" class="headerlink" title="计费模式"></a>计费模式</h2><p>商用RocketMq主要分为下面几个部分：</p>
<ul>
<li>主系列：标准版、专业版、铂金版</li>
<li>子系列：单节点版、集群版</li>
</ul>
<p>个人最后是白嫖了公司的商用集群版RocketMq进行实验，计费的方式分为<strong>包年包月</strong>和<strong>按量付费</strong>，前者适用于流量比较大并且固定的情况，使用套餐比较划算，而后者按需付费则适用于RocketMq使用较少（或者不稳定）或者调用量较少的情况，个人学习也比较推荐按量付费模式，比包年包月划算很多。</p>
<p>如果出现欠费，在实例停服7天后两种付费模式均会清除所有的RocketMq数据，而日常欠费则会被自动停用，嘛，和电话卡欠费差不多的道理，这里就不过多拓展了。如果要退订商用RocketMq的服务。按量付费可以随时退订，包年包月也会根据天数进行退订。</p>
<blockquote>
<p>PS：看完这一整套计费模式下来，发现还是挺良心的。让我没想到的是包年包月居然可以计算未使用的天数返还，这一点挺不错的，用起来不用担心被套进去。</p>
</blockquote>
<h2 id="接入方式"><a href="#接入方式" class="headerlink" title="接入方式"></a>接入方式</h2><p>商用RocketMq的接入步骤如下：</p>
<p><img src="https://adong-picture.oss-cn-shenzhen.aliyuncs.com/adong/20221229175755.png" alt=""></p>
<p>官方在快速入门中提供了创建资源的解释视频：<span class="exturl" data-url="aHR0cHM6Ly9oZWxwLmFsaXl1bi5jb20vZG9jdW1lbnRfZGV0YWlsLzQ0MTkxNC5odG1s" title="https://help.aliyun.com/document_detail/441914.html">https://help.aliyun.com/document_detail/441914.html<i class="fa fa-external-link"></i></span>。有两个点需要注意，第一个点是RAM用户必须要进行账户授权才能正常使用，第二个点是对外访问方式设置，消息队列RocketMQ版支持VPC访问和公网访问。资源创建完成之后就可以使用官方提供的SDK收发消息，在使用之前需要注意下面这些前提：</p>
<ul>
<li><p><span class="exturl" data-url="aHR0cHM6Ly9oZWxwLmFsaXl1bi5jb20vZG9jdW1lbnRfZGV0YWlsLzQ0NDc1Mi5odG0jdGFzay0yMjMzNjgz" title="https://help.aliyun.com/document_detail/444752.htm#task-2233683">步骤二：创建资源<i class="fa fa-external-link"></i></span></p>
</li>
<li><p><span class="exturl" data-url="aHR0cHM6Ly93d3cuamV0YnJhaW5zLmNvbS9pZGVhLz9zcG09YTJjNGcuMTExODY2MjMuMC4wLjZiMWMyYjM2MXZhRUhG" title="https://www.jetbrains.com/idea/?spm=a2c4g.11186623.0.0.6b1c2b361vaEHF">安装IDEA<i class="fa fa-external-link"></i></span></p>
<p>您可以使用IntelliJ IDEA或者Eclipse，本文以IntelliJ IDEA Ultimate为例。</p>
</li>
<li><p><span class="exturl" data-url="aHR0cHM6Ly93d3cub3JhY2xlLmNvbS9qYXZhL3RlY2hub2xvZ2llcy9kb3dubG9hZHMv" title="https://www.oracle.com/java/technologies/downloads/">安装1.8或以上版本JDK<i class="fa fa-external-link"></i></span></p>
</li>
<li><p><span class="exturl" data-url="aHR0cHM6Ly9tYXZlbi5hcGFjaGUub3JnL2Rvd25sb2FkLmNnaT9zcG09YTJjNGcuMTExODY2MjMuMC4wLjMyYjk0ZmViZndha2VCJmZpbGU9ZG93bmxvYWQuY2dp" title="https://maven.apache.org/download.cgi?spm=a2c4g.11186623.0.0.32b94febfwakeB&file=download.cgi">安装2.5或以上版本Maven<i class="fa fa-external-link"></i></span></p>
</li>
</ul>
<p>之后便是在项目中引入JAVA依赖和复制粘贴模板代码验证。</p>
<h2 id="消息队列模型"><a href="#消息队列模型" class="headerlink" title="消息队列模型"></a>消息队列模型</h2><p>商用RocketMq的基本使用逻辑如下：</p>
<p><img src="https://adong-picture.oss-cn-shenzhen.aliyuncs.com/adong/20221229182359.png" alt=""></p>
<p>按照消息类型，商用RocketMq提供了下面的消息类型：</p>
<ul>
<li><span class="exturl" data-url="aHR0cHM6Ly9oZWxwLmFsaXl1bi5jb20vZG9jdW1lbnRfZGV0YWlsLzQ0MDI0MC5odG1sP3NwbT1hMmM0Zy4xMTE4NjYyMy4wLjAuNTljMzQ2NmJsOGpFbk0=" title="https://help.aliyun.com/document_detail/440240.html?spm=a2c4g.11186623.0.0.59c3466bl8jEnM">普通消息<i class="fa fa-external-link"></i></span></li>
<li><span class="exturl" data-url="aHR0cHM6Ly9oZWxwLmFsaXl1bi5jb20vZG9jdW1lbnRfZGV0YWlsLzQ0MDI0MS5odG1sP3NwbT1hMmM0Zy4xMTE4NjYyMy4wLjAuMTFmNDI4NGNmRUpIdFU=" title="https://help.aliyun.com/document_detail/440241.html?spm=a2c4g.11186623.0.0.11f4284cfEJHtU">顺序消息<i class="fa fa-external-link"></i></span></li>
<li><span class="exturl" data-url="aHR0cHM6Ly9oZWxwLmFsaXl1bi5jb20vZG9jdW1lbnRfZGV0YWlsLzQ0MDI0Mi5odG1sP3NwbT1hMmM0Zy4xMTE4NjYyMy4wLjAuZWQ4OTRkMzg1ZFdmR0M=" title="https://help.aliyun.com/document_detail/440242.html?spm=a2c4g.11186623.0.0.ed894d385dWfGC">定时/延时消息<i class="fa fa-external-link"></i></span></li>
<li><span class="exturl" data-url="aHR0cHM6Ly9oZWxwLmFsaXl1bi5jb20vZG9jdW1lbnRfZGV0YWlsLzQ0MDI0NC5odG1sP3NwbT1hMmM0Zy4xMTE4NjYyMy4wLjAuMTFmNDRhZDRlTjFpenk=" title="https://help.aliyun.com/document_detail/440244.html?spm=a2c4g.11186623.0.0.11f44ad4eN1izy">事务消息<i class="fa fa-external-link"></i></span></li>
</ul>
<h2 id="SDK接入参考"><a href="#SDK接入参考" class="headerlink" title="SDK接入参考"></a>SDK接入参考</h2><p>SDK接入的主页链接如下：</p>
<p><span class="exturl" data-url="aHR0cHM6Ly9oZWxwLmFsaXl1bi5jb20vZG9jdW1lbnRfZGV0YWlsLzY5MTExLmh0bWw=" title="https://help.aliyun.com/document_detail/69111.html">https://help.aliyun.com/document_detail/69111.html<i class="fa fa-external-link"></i></span></p>
<p>主要介绍了下面几个点，实际参考第二个推荐的接入方式即可。</p>
<ul>
<li><span class="exturl" data-url="aHR0cHM6Ly9oZWxwLmFsaXl1bi5jb20vZG9jdW1lbnRfZGV0YWlsLzEyNDY5My5odG1s" title="https://help.aliyun.com/document_detail/124693.html">SDK参考概述<i class="fa fa-external-link"></i></span></li>
<li><a href="https://help.aliyun.com/document_detail/29544.html" target="_blank" rel="noopener"><strong>商业版TCP协议SDK（推荐）</strong></a></li>
<li><span class="exturl" data-url="aHR0cHM6Ly9oZWxwLmFsaXl1bi5jb20vZG9jdW1lbnRfZGV0YWlsLzE0MTc3OC5odG1s" title="https://help.aliyun.com/document_detail/141778.html">商业版HTTP协议SDK（多语言推荐）<i class="fa fa-external-link"></i></span></li>
<li><span class="exturl" data-url="aHR0cHM6Ly9oZWxwLmFsaXl1bi5jb20vZG9jdW1lbnRfZGV0YWlsLzEyODYwMC5odG1s" title="https://help.aliyun.com/document_detail/128600.html">社区版TCP协议SDK（仅供开源用户上云使用）<i class="fa fa-external-link"></i></span></li>
<li><span class="exturl" data-url="aHR0cHM6Ly9oZWxwLmFsaXl1bi5jb20vZG9jdW1lbnRfZGV0YWlsLzE2NDU2Ny5odG1s" title="https://help.aliyun.com/document_detail/164567.html">订阅消息常见问题<i class="fa fa-external-link"></i></span></li>
</ul>
<p>对于JAVA应用程序建议使用TCP协议SDK，集成的方式也比较简单，为了方便理解，也可以通过以下链接获取相关Demo，里面都有详细的注释解释：</p>
<ul>
<li><span class="exturl" data-url="aHR0cHM6Ly9naXRodWIuY29tL0FsaXdhcmVNUS9tcS1kZW1vL3RyZWUvbWFzdGVyL3NwcmluZy9qYXZhLXNwcmluZy1kZW1v" title="https://github.com/AliwareMQ/mq-demo/tree/master/spring/java-spring-demo">spring/java-spring-demo<i class="fa fa-external-link"></i></span></li>
<li><span class="exturl" data-url="aHR0cHM6Ly9naXRodWIuY29tL0FsaXdhcmVNUS9tcS1kZW1vL3RyZWUvbWFzdGVyL3NwcmluZ2Jvb3QvamF2YS1zcHJpbmdib290LWRlbW8=" title="https://github.com/AliwareMQ/mq-demo/tree/master/springboot/java-springboot-demo">springboot/java-springboot-demo<i class="fa fa-external-link"></i></span></li>
</ul>
<p>如果是HTTP协议的SDK，则使用下面的依赖包：</p>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>com.aliyun.mq<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>mq-http-sdk<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">    <span class="comment">&lt;!--以下版本号请替换为Java SDK的最新版本号--&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">version</span>&gt;</span>1.0.3.2<span class="tag">&lt;/<span class="name">version</span>&gt;</span> </span><br><span class="line">    <span class="tag">&lt;<span class="name">classifier</span>&gt;</span>jar-with-dependencies<span class="tag">&lt;/<span class="name">classifier</span>&gt;</span></span><br><span class="line"><span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br></pre></td></tr></table></figure>

<p>如果是TCP协议的接入方式，则使用下面的依赖包：</p>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>com.aliyun.openservices<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>ons-client<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">    <span class="comment">&lt;!--以下版本号请替换为Java SDK的最新版本号--&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">version</span>&gt;</span>1.8.8.1.Final<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line"><span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br></pre></td></tr></table></figure>



<h3 id="接入细节说明"><a href="#接入细节说明" class="headerlink" title="接入细节说明"></a>接入细节说明</h3><p>个人在实验的时候发现HTTP协议的版本集成SDK更像是给了一套API工具包，好处是可以不需要依赖Spring框架等单独使用，但是最大的缺点也是这里，很难用于框架当中，如果要实现自动接收消息并且处理，需要依靠<strong>手动实现定时任务</strong>拉取消息实现自动消费，这一点也比较蛋疼。</p>
<p>这里列举 <strong>HTTP的SDK</strong>的发送和接受代码，首先是生产者的模板代码，这些代码和开源的RocketMq使用类似：</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">// 省略大量代码。。。</span></span><br><span class="line"><span class="comment">// 循环发送4条消息。</span></span><br><span class="line">            <span class="keyword">for</span> (<span class="keyword">int</span> i = <span class="number">0</span>; i &lt; <span class="number">4</span>; i++) &#123;</span><br><span class="line">                TopicMessage pubMsg;        <span class="comment">// 普通消息。</span></span><br><span class="line">                pubMsg = <span class="keyword">new</span> TopicMessage(</span><br><span class="line">                <span class="comment">// 消息内容。</span></span><br><span class="line">                <span class="string">"hello mq!"</span>.getBytes(),</span><br><span class="line">                <span class="comment">// 消息标签。</span></span><br><span class="line">                <span class="string">"A"</span></span><br><span class="line">                );</span><br><span class="line">                <span class="comment">// 设置消息的自定义属性。</span></span><br><span class="line">                pubMsg.getProperties().put(<span class="string">"a"</span>, String.valueOf(i));</span><br><span class="line">                <span class="comment">// 设置消息的Key。</span></span><br><span class="line">                pubMsg.setMessageKey(<span class="string">"MessageKey"</span>);</span><br><span class="line">               </span><br><span class="line">            <span class="comment">// 同步发送消息，只要不抛异常就是成功。</span></span><br><span class="line">            TopicMessage pubResultMsg = producer.publishMessage(pubMsg);</span><br><span class="line"></span><br><span class="line">            <span class="comment">// 同步发送消息，只要不抛异常就是成功。</span></span><br><span class="line">            System.out.println(<span class="keyword">new</span> Date() + <span class="string">" Send mq message success. Topic is:"</span> + topic + <span class="string">", msgId is: "</span> + pubResultMsg.getMessageId()</span><br><span class="line">                        + <span class="string">", bodyMD5 is: "</span> + pubResultMsg.getMessageBodyMD5());</span><br><span class="line">            &#125;</span><br><span class="line">        &#125; <span class="keyword">catch</span> (Throwable e) &#123;</span><br><span class="line">            <span class="comment">// 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理。</span></span><br><span class="line">            System.out.println(<span class="keyword">new</span> Date() + <span class="string">" Send mq message failed. Topic is:"</span> + topic);</span><br><span class="line">            e.printStackTrace();</span><br><span class="line">        &#125;</span><br><span class="line"><span class="comment">// 省略大量代码。。。</span></span><br></pre></td></tr></table></figure>

<p>而在接收方稍微麻烦一些，官方的案例是使用 <strong>死循环</strong>不断检查Broker是否有积压消息，如果有则通过<strong>主动拉取消息</strong>的模式去拉取消息。</p>
<blockquote>
<p> 实现多线程等待的效果可以写入到Thread继承类的Run方法中。</p>
</blockquote>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br></pre></td><td class="code"><pre><span class="line"> <span class="comment">// 在当前线程循环消费消息，建议多开个几个线程并发消费消息。</span></span><br><span class="line">        <span class="keyword">do</span> &#123;</span><br><span class="line">            </span><br><span class="line">            <span class="comment">// 省略大量代码</span></span><br><span class="line"></span><br><span class="line"><span class="comment">// 处理业务逻辑。</span></span><br><span class="line">            <span class="keyword">for</span> (Message message : messages) &#123;</span><br><span class="line">                System.out.println(<span class="string">"Receive message: "</span> + message);</span><br><span class="line">            &#125;</span><br><span class="line"></span><br><span class="line">            <span class="comment">// 消息重试时间到达前若不确认消息消费成功，则消息会被重复消费。</span></span><br><span class="line">            <span class="comment">// 消息句柄有时间戳，同一条消息每次消费的时间戳都不一样。</span></span><br><span class="line">            &#123;</span><br><span class="line">                List&lt;String&gt; handles = <span class="keyword">new</span> ArrayList&lt;String&gt;();</span><br><span class="line">                <span class="keyword">for</span> (Message message : messages) &#123;</span><br><span class="line">                    handles.add(message.getReceiptHandle());</span><br><span class="line">                &#125;</span><br><span class="line"></span><br><span class="line">                <span class="keyword">try</span> &#123;</span><br><span class="line">                    consumer.ackMessage(handles);</span><br><span class="line">                &#125; <span class="keyword">catch</span> (Throwable e) &#123;</span><br><span class="line">                    <span class="comment">// 某些消息的句柄可能超时，会导致消息消费状态确认不成功。</span></span><br><span class="line">                    <span class="keyword">if</span> (e <span class="keyword">instanceof</span> AckMessageException) &#123;</span><br><span class="line">                        AckMessageException errors = (AckMessageException) e;</span><br><span class="line">                        System.out.println(<span class="string">"Ack message fail, requestId is:"</span> + errors.getRequestId() + <span class="string">", fail handles:"</span>);</span><br><span class="line">                        <span class="keyword">if</span> (errors.getErrorMessages() != <span class="keyword">null</span>) &#123;</span><br><span class="line">                            <span class="keyword">for</span> (String errorHandle :errors.getErrorMessages().keySet()) &#123;</span><br><span class="line">                                System.out.println(<span class="string">"Handle:"</span> + errorHandle + <span class="string">", ErrorCode:"</span> + errors.getErrorMessages().get(errorHandle).getErrorCode()</span><br><span class="line">                                        + <span class="string">", ErrorMsg:"</span> + errors.getErrorMessages().get(errorHandle).getErrorMessage());</span><br><span class="line">                            &#125;</span><br><span class="line">                        &#125;</span><br><span class="line">                        <span class="keyword">continue</span>;</span><br><span class="line">                    &#125;</span><br><span class="line">                    e.printStackTrace();</span><br><span class="line">                &#125;</span><br><span class="line">            &#125;</span><br><span class="line">       &#125; <span class="keyword">while</span> (<span class="keyword">true</span>);</span><br><span class="line">     <span class="comment">// 省略大量代码</span></span><br></pre></td></tr></table></figure>



<h1 id="写在最后"><a href="#写在最后" class="headerlink" title="写在最后"></a>写在最后</h1><p>算是一次个人兼容的笔记。还是要吐槽商用的RocketMq集成真的挺无语的，不过阿里的东西懂的都懂。</p>
<script type="text/javascript" src="https://cdn.jsdelivr.net/npm/kity@2.0.4/dist/kity.min.js"></script><script type="text/javascript" src="https://cdn.jsdelivr.net/npm/kityminder-core@1.4.50/dist/kityminder.core.min.js"></script><script defer="true" type="text/javascript" src="https://cdn.jsdelivr.net/npm/hexo-simple-mindmap@0.2.0/dist/mindmap.min.js"></script><link rel="stylesheet" type="text/css" href="https://cdn.jsdelivr.net/npm/hexo-simple-mindmap@0.2.0/dist/mindmap.min.css">
    </div>

    
    
    
        

<div>
<ul class="post-copyright">
  <li class="post-copyright-author">
    <strong>本文作者： </strong>lazytime
  </li>
  <li class="post-copyright-link">
    <strong>本文链接：</strong>
    <a href="https://whitestore.top/2023/03/11/shangyong/" title="【RocketMq】商用RocketMq和开源RocketMq的兼容问题解决方案">https://whitestore.top/2023/03/11/shangyong/</a>
  </li>
  <li class="post-copyright-license">
    <strong>版权声明： </strong>本博客所有文章除特别声明外，均采用 <span class="exturl" data-url="aHR0cHM6Ly9jcmVhdGl2ZWNvbW1vbnMub3JnL2xpY2Vuc2VzL2J5LW5jLzQuMC96aC1DTg=="><i class="fa fa-fw fa-creative-commons"></i>BY-NC</span> 许可协议。转载请注明出处！
  </li>
</ul>
</div>


      <footer class="post-footer">
          <div class="post-tags">
              <a href="/tags/RocketMq/" rel="tag"># RocketMq</a>
          </div>

        


        
    <div class="post-nav">
      <div class="post-nav-item">
    <a href="/2023/03/11/codingparctice/" rel="prev" title="【Java】Best coding practices every java developer should">
      <i class="fa fa-chevron-left"></i> 【Java】Best coding practices every java developer should
    </a></div>
      <div class="post-nav-item">
    <a href="/2023/03/11/rocketjianhua/" rel="next" title="【RocketMq】使用Idea部署RocketMq 源代码（4.9.4）">
      【RocketMq】使用Idea部署RocketMq 源代码（4.9.4） <i class="fa fa-chevron-right"></i>
    </a></div>
    </div>
      </footer>
    
  </article>
  
  
  



          </div>
          
    <div class="comments" id="valine-comments"></div>

<script>
  window.addEventListener('tabs:register', () => {
    let { activeClass } = CONFIG.comments;
    if (CONFIG.comments.storage) {
      activeClass = localStorage.getItem('comments_active') || activeClass;
    }
    if (activeClass) {
      let activeTab = document.querySelector(`a[href="#comment-${activeClass}"]`);
      if (activeTab) {
        activeTab.click();
      }
    }
  });
  if (CONFIG.comments.storage) {
    window.addEventListener('tabs:click', event => {
      if (!event.target.matches('.tabs-comment .tab-content .tab-pane')) return;
      let commentClass = event.target.classList[1];
      localStorage.setItem('comments_active', commentClass);
    });
  }
</script>

        </div>
          
  
  <div class="toggle sidebar-toggle">
    <span class="toggle-line toggle-line-first"></span>
    <span class="toggle-line toggle-line-middle"></span>
    <span class="toggle-line toggle-line-last"></span>
  </div>

  <aside class="sidebar">
    <div class="sidebar-inner">

      <ul class="sidebar-nav motion-element">
        <li class="sidebar-nav-toc">
          文章目录
        </li>
        <li class="sidebar-nav-overview">
          站点概览
        </li>
      </ul>

      <!--noindex-->
      <div class="post-toc-wrap sidebar-panel">
          <div class="post-toc motion-element"><ol class="nav"><li class="nav-item nav-level-1"><a class="nav-link" href="#引言"><span class="nav-number">1.</span> <span class="nav-text">引言</span></a></li><li class="nav-item nav-level-1"><a class="nav-link" href="#兼容关键点"><span class="nav-number">2.</span> <span class="nav-text">兼容关键点</span></a></li><li class="nav-item nav-level-1"><a class="nav-link" href="#SpringBoot2-X-兼容"><span class="nav-number">3.</span> <span class="nav-text">SpringBoot2.X 兼容</span></a><ol class="nav-child"><li class="nav-item nav-level-2"><a class="nav-link" href="#SpringBoot项目兼容"><span class="nav-number">3.1.</span> <span class="nav-text">SpringBoot项目兼容</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#开源版本"><span class="nav-number">3.1.1.</span> <span class="nav-text">开源版本</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#商用版本"><span class="nav-number">3.1.2.</span> <span class="nav-text">商用版本</span></a></li></ol></li><li class="nav-item nav-level-2"><a class="nav-link" href="#SpringBoot项目兼容小结"><span class="nav-number">3.2.</span> <span class="nav-text">SpringBoot项目兼容小结</span></a></li><li class="nav-item nav-level-2"><a class="nav-link" href="#Spring-Cloud-Alibaba项目兼容"><span class="nav-number">3.3.</span> <span class="nav-text">Spring Cloud Alibaba项目兼容</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#开源版本-1"><span class="nav-number">3.3.1.</span> <span class="nav-text">开源版本</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#商用版本-1"><span class="nav-number">3.3.2.</span> <span class="nav-text">商用版本</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#Spring-Cloud-Alibaba项目兼容小结"><span class="nav-number">3.3.3.</span> <span class="nav-text">Spring Cloud Alibaba项目兼容小结</span></a></li></ol></li></ol></li><li class="nav-item nav-level-1"><a class="nav-link" href="#阿里云商用RocketMq简单介绍（4-X版本）"><span class="nav-number">4.</span> <span class="nav-text">阿里云商用RocketMq简单介绍（4.X版本）</span></a><ol class="nav-child"><li class="nav-item nav-level-2"><a class="nav-link" href="#计费模式"><span class="nav-number">4.1.</span> <span class="nav-text">计费模式</span></a></li><li class="nav-item nav-level-2"><a class="nav-link" href="#接入方式"><span class="nav-number">4.2.</span> <span class="nav-text">接入方式</span></a></li><li class="nav-item nav-level-2"><a class="nav-link" href="#消息队列模型"><span class="nav-number">4.3.</span> <span class="nav-text">消息队列模型</span></a></li><li class="nav-item nav-level-2"><a class="nav-link" href="#SDK接入参考"><span class="nav-number">4.4.</span> <span class="nav-text">SDK接入参考</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#接入细节说明"><span class="nav-number">4.4.1.</span> <span class="nav-text">接入细节说明</span></a></li></ol></li></ol></li><li class="nav-item nav-level-1"><a class="nav-link" href="#写在最后"><span class="nav-number">5.</span> <span class="nav-text">写在最后</span></a></li></ol></div>
      </div>
      <!--/noindex-->

      <div class="site-overview-wrap sidebar-panel">
        <div class="site-author motion-element" itemprop="author" itemscope itemtype="http://schema.org/Person">
  <p class="site-author-name" itemprop="name">阿东</p>
  <div class="site-description" itemprop="description">随遇而安</div>
</div>
<div class="site-state-wrap motion-element">
  <nav class="site-state">
      <div class="site-state-item site-state-posts">
          <a href="/archives/">
        
          <span class="site-state-item-count">239</span>
          <span class="site-state-item-name">日志</span>
        </a>
      </div>
      <div class="site-state-item site-state-categories">
            <a href="/categories/">
          
        <span class="site-state-item-count">36</span>
        <span class="site-state-item-name">分类</span></a>
      </div>
      <div class="site-state-item site-state-tags">
            <a href="/tags/">
          
        <span class="site-state-item-count">37</span>
        <span class="site-state-item-name">标签</span></a>
      </div>
  </nav>
</div>
  <div class="links-of-author motion-element">
      <span class="links-of-author-item">
        <span class="exturl" data-url="aHR0cHM6Ly9naXRodWIuY29tL2xhenlUaW1lcw==" title="GitHub → https:&#x2F;&#x2F;github.com&#x2F;lazyTimes"><i class="fa fa-fw fa-github"></i>GitHub</span>
      </span>
      <span class="links-of-author-item">
        <span class="exturl" data-url="bWFpbHRvOjEwOTc0ODM1MDhAcXEuY29t" title="E-Mail → mailto:1097483508@qq.com"><i class="fa fa-fw fa-envelope"></i>E-Mail</span>
      </span>
  </div>


  <div class="links-of-blogroll motion-element">
    <div class="links-of-blogroll-title">
      <i class="fa fa-fw fa-link"></i>
      友情链接
    </div>
    <ul class="links-of-blogroll-list">
        <li class="links-of-blogroll-item">
          <span class="exturl" data-url="aHR0cHM6Ly93d3cuNTJwb2ppZS5jbi9ob21lLnBocD9tb2Q9c3BhY2UmdWlkPTE0OTc3MTgmZG89dGhyZWFkJnZpZXc9bWUmZnJvbT1zcGFjZQ==" title="https:&#x2F;&#x2F;www.52pojie.cn&#x2F;home.php?mod&#x3D;space&amp;uid&#x3D;1497718&amp;do&#x3D;thread&amp;view&#x3D;me&amp;from&#x3D;space">吾爱破解</span>
        </li>
        <li class="links-of-blogroll-item">
          <span class="exturl" data-url="aHR0cHM6Ly9qdWVqaW4uaW0vdXNlci8yOTk5MTIzNDUyNjI2MzY2" title="https:&#x2F;&#x2F;juejin.im&#x2F;user&#x2F;2999123452626366">掘金</span>
        </li>
        <li class="links-of-blogroll-item">
          <span class="exturl" data-url="aHR0cHM6Ly9zZWdtZW50ZmF1bHQuY29tL3UvbGF6eXRpbWVz" title="https:&#x2F;&#x2F;segmentfault.com&#x2F;u&#x2F;lazytimes">思否</span>
        </li>
    </ul>
  </div>

      </div>

      <div class="wechat_OA">
        <span>欢迎关注我的公众号</span>
        <br>
          <!-- 这里添加你的二维码图片 -->
        <img src ="https://adong-picture.oss-cn-shenzhen.aliyuncs.com/adong/wechat_channel.jpg">
      </div>
        <div class="back-to-top motion-element">
          <i class="fa fa-arrow-up"></i>
          <span>0%</span>
        </div>

    </div>
  </aside>
  <div id="sidebar-dimmer"></div>


      </div>
    </main>

    <footer class="footer">
      <div class="footer-inner">
        

        

<div class="copyright">
  
  &copy; 
  <span itemprop="copyrightYear">2023</span>
  <span class="with-love">
    <i class="fa fa-user"></i>
  </span>
  <span class="author" itemprop="copyrightHolder">阿东</span>
    <span class="post-meta-divider">|</span>
    <span class="post-meta-item-icon">
      <i class="fa fa-area-chart"></i>
    </span>
      <span class="post-meta-item-text">站点总字数：</span>
    <span title="站点总字数">2m</span>
    <span class="post-meta-divider">|</span>
    <span class="post-meta-item-icon">
      <i class="fa fa-coffee"></i>
    </span>
      <span class="post-meta-item-text">站点阅读时长 &asymp;</span>
    <span title="站点阅读时长">29:50</span>
</div>
  <div class="powered-by">由 <span class="exturl theme-link" data-url="aHR0cHM6Ly9oZXhvLmlv">Hexo</span> & <span class="exturl theme-link" data-url="aHR0cHM6Ly90aGVtZS1uZXh0Lm9yZw==">NexT.Gemini</span> 强力驱动
  </div>

        
<div class="busuanzi-count">
  <script async src="https://busuanzi.ibruce.info/busuanzi/2.3/busuanzi.pure.mini.js"></script>
    <span class="post-meta-item" id="busuanzi_container_site_uv" style="display: none;">
      <span class="post-meta-item-icon">
        <i class="fa fa-user"></i>
      </span>
      <span class="site-uv" title="总访客量">
        <span id="busuanzi_value_site_uv"></span>
      </span>
    </span>
    <span class="post-meta-divider">|</span>
    <span class="post-meta-item" id="busuanzi_container_site_pv" style="display: none;">
      <span class="post-meta-item-icon">
        <i class="fa fa-eye"></i>
      </span>
      <span class="site-pv" title="总访问量">
        <span id="busuanzi_value_site_pv"></span>
      </span>
    </span>
</div>








      </div>
    </footer>
  </div>

  
  <script src="/lib/anime.min.js"></script>
  <script src="//cdn.jsdelivr.net/npm/jquery@3/dist/jquery.min.js"></script>
  <script src="//cdn.jsdelivr.net/gh/fancyapps/fancybox@3/dist/jquery.fancybox.min.js"></script>
  <script src="/lib/velocity/velocity.min.js"></script>
  <script src="/lib/velocity/velocity.ui.min.js"></script>

<script src="/js/utils.js"></script>

<script src="/js/motion.js"></script>


<script src="/js/schemes/pisces.js"></script>


<script src="/js/next-boot.js"></script>




  




  
<script src="/js/local-search.js"></script>













  

  


<script>
NexT.utils.loadComments(document.querySelector('#valine-comments'), () => {
  NexT.utils.getScript('//unpkg.com/valine/dist/Valine.min.js', () => {
    var GUEST = ['nick', 'mail', 'link'];
    var guest = 'nick,mail,link';
    guest = guest.split(',').filter(item => {
      return GUEST.includes(item);
    });
    new Valine({
      el         : '#valine-comments',
      verify     : false,
      notify     : true,
      appId      : 'qMUpEEvBgXaMDD1b0ftgi9xr-gzGzoHsz',
      appKey     : 'UCdfT4Rfih6MO6y8DI4fstf6',
      placeholder: "Just go go",
      avatar     : 'mm',
      meta       : guest,
      pageSize   : '10' || 10,
      visitor    : false,
      lang       : 'zh-CN' || 'zh-cn',
      path       : location.pathname,
      recordIP   : false,
      serverURLs : ''
    });
  }, window.Valine);
});
</script>

</body>
</html>
